From 5df0e5283d4f67eb3e6b2f210b3f89ff1768fb6b Mon Sep 17 00:00:00 2001 From: Chalindu Kodikara Date: Fri, 10 Jul 2026 15:09:31 +0530 Subject: [PATCH 1/2] feat: add events support for aws logs module Signed-off-by: Chalindu Kodikara --- observability-logs-aws-cloudwatch/Makefile | 7 + observability-logs-aws-cloudwatch/README.md | 242 +++++++++++++++--- observability-logs-aws-cloudwatch/VERSION | 2 +- .../helm/templates/_helpers.tpl | 16 ++ .../helm/templates/adapter/configmap.yaml | 1 + .../helm/templates/cloudwatch-setup/job.yaml | 8 + .../helm/values.yaml | 21 +- .../init/setup-cloudwatch.sh | 49 +++- .../internal/api/gen/models.gen.go | 164 +++++++++++- .../internal/api/gen/server.gen.go | 204 ++++++++++++--- .../internal/cloudwatch/client.go | 4 + .../internal/cloudwatch/event_labels.go | 71 +++++ .../internal/cloudwatch/event_queries.go | 117 +++++++++ .../internal/cloudwatch/event_queries_test.go | 119 +++++++++ .../internal/cloudwatch/event_types.go | 54 ++++ .../internal/cloudwatch/events.go | 91 +++++++ .../internal/cloudwatch/events_test.go | 107 ++++++++ .../internal/config.go | 20 +- .../internal/config_test.go | 29 +++ .../internal/event_handlers_test.go | 225 ++++++++++++++++ .../internal/handlers.go | 168 ++++++++++++ .../internal/handlers_test.go | 23 ++ observability-logs-aws-cloudwatch/main.go | 2 + 23 files changed, 1645 insertions(+), 99 deletions(-) create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/event_labels.go create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries.go create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries_test.go create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/event_types.go create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/events.go create mode 100644 observability-logs-aws-cloudwatch/internal/cloudwatch/events_test.go create mode 100644 observability-logs-aws-cloudwatch/internal/event_handlers_test.go diff --git a/observability-logs-aws-cloudwatch/Makefile b/observability-logs-aws-cloudwatch/Makefile index d799813d..23b84c6a 100644 --- a/observability-logs-aws-cloudwatch/Makefile +++ b/observability-logs-aws-cloudwatch/Makefile @@ -2,6 +2,13 @@ CFG_DIR := internal/api OAPI_CODEGEN_VERSION ?= v2.5.1 # Pinned to an immutable commit SHA so `make openapi-codegen` is reproducible. # Bump the SHA when the upstream spec changes. +# +# NOTE: the events API (`POST /api/v1/events/query`, EventsQueryRequest/Response, +# EventEntry) must be present in this upstream spec before regenerating. Until that +# lands upstream, the generated `internal/api/gen/*` was produced from the local +# TASKS/logs-event-publishing/openapi.yaml — regenerating against a spec that lacks +# the events path will strip the events code and break the build. Override with e.g. +# `make openapi-codegen SPEC=../TASKS/logs-event-publishing/openapi.yaml` in the meantime. SPEC := https://raw.githubusercontent.com/openchoreo/openchoreo/main/openapi/observability-logs-adapter-api.yaml .PHONY: oapi-codegen-install openapi-codegen unit-test diff --git a/observability-logs-aws-cloudwatch/README.md b/observability-logs-aws-cloudwatch/README.md index 3c0fd580..187a9117 100644 --- a/observability-logs-aws-cloudwatch/README.md +++ b/observability-logs-aws-cloudwatch/README.md @@ -12,17 +12,18 @@ This module supports both: ## Table of contents 1. [Architecture](#architecture) -2. [Choose a deployment topology](#choose-a-deployment-topology) -3. [Prerequisites](#prerequisites) -4. [IAM permissions](#iam-permissions) -5. [Installation on EKS with Pod Identity](#installation-on-eks-with-pod-identity) -6. [Installation on non-EKS clusters with static credentials](#installation-on-non-eks-clusters-with-static-credentials) -7. [Log alerting](#log-alerting) -8. [Expose the alert webhook through EventBridge](#expose-the-alert-webhook-through-eventbridge) -9. [Shared webhook secret](#shared-webhook-secret) -10. [Troubleshooting](#troubleshooting) -11. [Configuration reference](#configuration-reference) -12. [Compatibility](#compatibility) +2. [Kubernetes events](#kubernetes-events) +3. [Choose a deployment topology](#choose-a-deployment-topology) +4. [Prerequisites](#prerequisites) +5. [IAM permissions](#iam-permissions) +6. [Installation on EKS with Pod Identity](#installation-on-eks-with-pod-identity) +7. [Installation on non-EKS clusters with static credentials](#installation-on-non-eks-clusters-with-static-credentials) +8. [Log alerting](#log-alerting) +9. [Expose the alert webhook through EventBridge](#expose-the-alert-webhook-through-eventbridge) +10. [Shared webhook secret](#shared-webhook-secret) +11. [Troubleshooting](#troubleshooting) +12. [Configuration reference](#configuration-reference) +13. [Compatibility](#compatibility) ## Architecture @@ -51,9 +52,18 @@ Each log record includes Kubernetes metadata such as: All participating clusters write to the same configured application log group. +Kubernetes **events** are an optional capability written to a separate log group: + +```text +/aws/containerinsights/events +``` + +See [Kubernetes events](#kubernetes-events) for how to publish and query them. + | Endpoint | Purpose | | --- | --- | | `POST /api/v1/logs/query` | Runs a CloudWatch Logs Insights query and filters logs by OpenChoreo scope labels. | +| `POST /api/v1/events/query` | Runs a CloudWatch Logs Insights query over the events log group, scoped by component or workflow. Returns an empty result when the events log group is absent or no events have been published. | | `POST /api/v1alpha1/alerts/rules` | Creates a CloudWatch Logs metric filter and CloudWatch metric alarm. | | `GET /api/v1alpha1/alerts/rules/{ruleName}` | Gets the alert rule identified by `{ruleName}`. | | `PUT /api/v1alpha1/alerts/rules/{ruleName}` | Updates the alert rule identified by `{ruleName}`. | @@ -62,6 +72,86 @@ All participating clusters write to the same configured application log group. | `GET /healthz` | Readiness check. Returns `200` once the adapter is ready. | | `GET /livez` | Liveness check. Does not call AWS, so transient AWS or DNS issues do not crash-loop the pod. | +## Kubernetes events + +Alongside container logs, the adapter can answer `POST /api/v1/events/query` with +Kubernetes events (Pod scheduling failures, image pull errors, `BackOff`, workflow +job lifecycle, and so on), scoped by component or workflow the same way logs are. + +Kubernetes event ingestion has two parts: + +1. **Ingestion** — the separate [`observability-events-otel-collector`](../observability-events-otel-collector) + chart collects events cluster-wide, enriches each one with the involved object's + OpenChoreo labels (via its `k8seventenrich` processor), and ships them to a + dedicated events log group using its bundled `awscloudwatchlogs` exporter. Like + Fluent Bit, it must run on every workflow plane and data plane clusters, and all clusters + write to the same shared events log group. +2. **Query** — this adapter reads that log group. The query endpoint is always + served. When the log group does not exist or the collector has not published + events, `POST /api/v1/events/query` returns `{ "events": [], "total": 0 }` + rather than an error, so events degrade gracefully. + +### Provision the events log group + +The setup Job provisions the events log group with its own retention by default +(30 days). Set `events.provisionLogGroup=false` only for log-only installs that +intentionally do not grant IAM permissions for the events log group: + +```yaml +events: + provisionLogGroup: true + retentionDays: 30 # one of CloudWatch's allowed retention values +``` + +The log group name is derived from the shared prefix as +`/events` (default `/aws/containerinsights/events`), mirroring +the application log group. The adapter picks it up automatically via the +`EVENTS_LOG_GROUP_NAME` config value. If you need an existing or non-standard +CloudWatch log group, set `events.logGroupName` and point the events collector's +`awscloudwatchlogs.log_group_name` at the same value. + +### Deploy the events collector + +Install `observability-events-otel-collector` on each workload cluster with its +CloudWatch exporter pointed at the same log group and region: + +```yaml +exporters: + awscloudwatchlogs: + region: "" + log_group_name: "/aws/containerinsights/events" + log_stream_name: "events" +pipelineExporters: + - awscloudwatchlogs +``` + +The collector's identity needs `logs:CreateLogStream` and `logs:PutLogEvents` on the +events log group (this chart's setup Job owns group creation and retention, so the +collector does not need `logs:CreateLogGroup`). See the +[events collector README](../observability-events-otel-collector/README.md#aws-cloudwatch-logs) +for the full backend configuration. + +### Query behavior + +Component-scoped event queries filter by the OpenChoreo labels copied from the +involved Kubernetes object: + +- `openchoreo.dev/namespace` +- `openchoreo.dev/project-uid` +- `openchoreo.dev/environment-uid` +- `openchoreo.dev/component-uid` + +Workflow-scoped event queries do not require project or component labels. Workflow +events are matched by Kubernetes namespace and workflow object name: + +- `attributes.k8s.namespace.name = workflows-` +- `resource.k8s.object.name = ` or `-...` + +When a workflow task is provided, the adapter also matches Argo's step and pod +naming patterns. For example, `taskName=checkout-source` matches both workflow-node +events whose `body` contains `checkout-source` and pod events named like +`-checkout-`. + ## Choose a deployment topology Choose the deployment topology first, then choose the AWS authentication model for each cluster. @@ -116,15 +206,16 @@ The CloudWatch adapter needs permissions for three paths: 1. Startup identity check. 2. Log query and metric-filter management. 3. CloudWatch alarm management. +4. Kubernetes event queries over the events log group. Use these policies based on the credential model: - **EKS Pod Identity or IRSA:** keep the adapter and ingestion policies separate and attach them to separate roles. This keeps each ServiceAccount least-privileged. -- **Static credentials:** use one IAM user and attach both the adapter policy and `CloudWatchAgentServerPolicy`, because the same Kubernetes Secret is shared by all components. +- **Static credentials:** use one IAM user and attach the adapter policy, setup Job policy, and `CloudWatchAgentServerPolicy`, because the same Kubernetes Secret is shared by all components. ### Fluent Bit IAM policy -The CloudWatch Agent and Fluent Bit need permission to write logs to CloudWatch. Attach the AWS-managed policy below: +The CloudWatch Agent and Fluent Bit need permission to write logs to CloudWatch. Attach the AWS-managed policy below: ```text CloudWatchAgentServerPolicy @@ -150,7 +241,7 @@ Replace: "Resource": "*" }, { - "Sid": "LogsScoped", + "Sid": "ApplicationLogsScoped", "Effect": "Allow", "Action": [ "logs:StartQuery", @@ -160,6 +251,14 @@ Replace: ], "Resource": "arn:aws:logs:::log-group:/aws/containerinsights/application:*" }, + { + "Sid": "EventLogsScoped", + "Effect": "Allow", + "Action": [ + "logs:StartQuery" + ], + "Resource": "arn:aws:logs:::log-group:/aws/containerinsights/events:*" + }, { "Sid": "LogsUnscoped", "Effect": "Allow", @@ -192,6 +291,7 @@ Notes: - `cloudwatch:UntagResource` is not required because the adapter does not remove tags. - CloudWatch alarm actions use `"Resource": "*"` because alarm ARNs are only known after the first `PutMetricAlarm` call. - Leave `adapter.alerting.alarmActionArns` empty when using EventBridge to forward alarm state-change events. +- If you override `global.applicationLogGroupName` or `events.logGroupName`, replace the default log group ARNs in the policy with the exact custom log group names. ### Setup Job IAM policy @@ -213,12 +313,18 @@ Replace: "logs:CreateLogGroup", "logs:PutRetentionPolicy" ], - "Resource": "arn:aws:logs:::log-group:/aws/containerinsights/application:*" + "Resource": [ + "arn:aws:logs:::log-group:/aws/containerinsights/application:*", + "arn:aws:logs:::log-group:/aws/containerinsights/events:*" + ] } ] } ``` +If you override `global.applicationLogGroupName` or `events.logGroupName`, replace +the default log group ARNs in the policy with the exact custom log group names. + ## Installation on EKS with Pod Identity This is the recommended installation path for EKS clusters. @@ -360,7 +466,7 @@ aws iam create-policy \ "Resource": "*" }, { - "Sid": "LogsScoped", + "Sid": "ApplicationLogsScoped", "Effect": "Allow", "Action": [ "logs:StartQuery", @@ -370,6 +476,14 @@ aws iam create-policy \ ], "Resource": "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/application:*" }, + { + "Sid": "EventLogsScoped", + "Effect": "Allow", + "Action": [ + "logs:StartQuery" + ], + "Resource": "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/events:*" + }, { "Sid": "LogsUnscoped", "Effect": "Allow", @@ -419,7 +533,10 @@ aws iam create-policy \ "logs:CreateLogGroup", "logs:PutRetentionPolicy" ], - "Resource": "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/application:*" + "Resource": [ + "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/application:*", + "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/events:*" + ] } ] } @@ -448,7 +565,7 @@ helm upgrade --install observability-logs-aws-cloudwatch \ oci://ghcr.io/openchoreo/helm-charts/observability-logs-aws-cloudwatch \ --create-namespace \ --namespace "$NS" \ - --version 0.2.1 \ + --version 0.3.0 \ --set amazon-cloudwatch-observability.region="$AWS_REGION" \ --set adapter.alerting.webhookAuth.enabled=true \ --set adapter.alerting.webhookAuth.sharedSecret="$WEBHOOK_SHARED_SECRET" @@ -463,7 +580,7 @@ helm upgrade --install observability-logs-aws-cloudwatch \ oci://ghcr.io/openchoreo/helm-charts/observability-logs-aws-cloudwatch \ --create-namespace \ --namespace "$NS" \ - --version 0.2.1 \ + --version 0.3.0 \ --set cloudWatchAgent.enabled=false \ --set setup.enabled=false \ --set amazon-cloudwatch-observability.region="$AWS_REGION" \ @@ -480,7 +597,7 @@ helm upgrade --install observability-logs-aws-cloudwatch \ oci://ghcr.io/openchoreo/helm-charts/observability-logs-aws-cloudwatch \ --create-namespace \ --namespace "$NS" \ - --version 0.2.1 \ + --version 0.3.0 \ --set amazon-cloudwatch-observability.region="$AWS_REGION" \ --set adapter.enabled=false ``` @@ -491,13 +608,14 @@ EKS Pod Identity links a Kubernetes ServiceAccount to an IAM role. Each associat #### Single-cluster topology -Create three Pod Identity associations on the EKS cluster, all in the `$NS` namespace: +Create Pod Identity associations on the EKS cluster, all in the `$NS` namespace: | ServiceAccount | Used by | IAM role to associate | | --- | --- | --- | | `logs-adapter-aws-cloudwatch` | Adapter queries, alerting CRUD, and webhook handling. | The role with the [Adapter IAM policy](#adapter-iam-policy) attached. | | `cloudwatch-setup` | Setup Job that creates log groups and applies retention. | The role with the [Setup Job IAM policy](#setup-job-iam-policy) and `CloudWatchAgentServerPolicy` attached. | | `cloudwatch-agent` | CloudWatch Agent and Fluent Bit DaemonSets. | The role with `CloudWatchAgentServerPolicy` attached. | +| `events-collector` | Optional Kubernetes event ingestion through `observability-events-otel-collector`. | The role with `CloudWatchAgentServerPolicy` attached. | #### Multi-cluster topology @@ -517,6 +635,7 @@ The `cloudwatch-setup` and `cloudwatch-agent` ServiceAccounts do not exist in th | --- | --- | | `cloudwatch-setup` | The role with the [Setup Job IAM policy](#setup-job-iam-policy) and `CloudWatchAgentServerPolicy` attached. | | `cloudwatch-agent` | The role with `CloudWatchAgentServerPolicy` attached. | +| `events-collector` | The role with `CloudWatchAgentServerPolicy` attached when `observability-events-otel-collector` is installed. | The `logs-adapter-aws-cloudwatch` ServiceAccount does not exist in these clusters because `adapter.enabled=false`. @@ -541,7 +660,7 @@ export ADAPTER_ROLE_ARN="arn:aws:iam::${AWS_ACCOUNT_ID}:role/OpenChoreoCloudWatc export INGESTION_ROLE_ARN="arn:aws:iam::${AWS_ACCOUNT_ID}:role/OpenChoreoCloudWatchLogsRoleForIngestion" ``` -**Single-cluster topology** — create all three associations: +**Single-cluster topology** — create associations for the adapter, setup Job, CloudWatch Agent, and optional events collector: ```bash export EKS_CLUSTER_NAME= @@ -563,6 +682,13 @@ aws eks create-pod-identity-association \ --namespace "$NS" \ --service-account cloudwatch-agent \ --role-arn "$INGESTION_ROLE_ARN" + +# Only needed when observability-events-otel-collector is installed in this cluster. +aws eks create-pod-identity-association \ + --cluster-name "$EKS_CLUSTER_NAME" \ + --namespace "$NS" \ + --service-account events-collector \ + --role-arn "$INGESTION_ROLE_ARN" ``` **Observability plane cluster** — adapter only: @@ -593,6 +719,13 @@ aws eks create-pod-identity-association \ --namespace "$NS" \ --service-account cloudwatch-agent \ --role-arn "$INGESTION_ROLE_ARN" + +# Only needed when observability-events-otel-collector is installed in this cluster. +aws eks create-pod-identity-association \ + --cluster-name "$EKS_CLUSTER_NAME" \ + --namespace "$NS" \ + --service-account events-collector \ + --role-arn "$INGESTION_ROLE_ARN" ``` ### Step 5 — Restart workloads on Each Cluster @@ -613,11 +746,15 @@ kubectl -n "$NS" rollout restart daemonset/cloudwatch-agent kubectl -n "$NS" rollout restart daemonset/fluent-bit kubectl -n "$NS" rollout restart deployment/logs-adapter-aws-cloudwatch +# If observability-events-otel-collector is installed or will be installed in this cluster: +kubectl -n "$NS" get deployment/events-collector >/dev/null 2>&1 && \ + kubectl -n "$NS" rollout restart deployment/events-collector + # Re-trigger the setup Helm hook (it ran once at install time) kubectl -n "$NS" delete job cloudwatch-setup-logs --ignore-not-found helm upgrade observability-logs-aws-cloudwatch \ oci://ghcr.io/openchoreo/helm-charts/observability-logs-aws-cloudwatch \ - --namespace "$NS" --version 0.2.1 --reuse-values + --namespace "$NS" --version 0.3.0 --reuse-values ``` **Observability plane cluster** — restart only the adapter: @@ -632,10 +769,14 @@ kubectl -n "$NS" rollout restart deployment/logs-adapter-aws-cloudwatch kubectl -n "$NS" rollout restart daemonset/cloudwatch-agent kubectl -n "$NS" rollout restart daemonset/fluent-bit +# If observability-events-otel-collector is installed or will be installed in this cluster: +kubectl -n "$NS" get deployment/events-collector >/dev/null 2>&1 && \ + kubectl -n "$NS" rollout restart deployment/events-collector + kubectl -n "$NS" delete job cloudwatch-setup-logs --ignore-not-found helm upgrade observability-logs-aws-cloudwatch \ oci://ghcr.io/openchoreo/helm-charts/observability-logs-aws-cloudwatch \ - --namespace "$NS" --version 0.2.1 --reuse-values + --namespace "$NS" --version 0.3.0 --reuse-values ``` #### Verify Pod Identity injection @@ -676,6 +817,7 @@ export AWS_SECRET_ACCESS_KEY= Create an IAM user and attach both: - The custom [Adapter IAM policy](#adapter-iam-policy). +- The custom [Setup Job IAM policy](#setup-job-iam-policy). - The AWS-managed `CloudWatchAgentServerPolicy`. Create access keys for this IAM user and export them as shown above. @@ -702,7 +844,7 @@ aws iam create-policy \ "Resource": "*" }, { - "Sid": "LogsScoped", + "Sid": "ApplicationLogsScoped", "Effect": "Allow", "Action": [ "logs:StartQuery", @@ -712,6 +854,14 @@ aws iam create-policy \ ], "Resource": "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/application:*" }, + { + "Sid": "EventLogsScoped", + "Effect": "Allow", + "Action": [ + "logs:StartQuery" + ], + "Resource": "arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:/aws/containerinsights/events:*" + }, { "Sid": "LogsUnscoped", "Effect": "Allow", @@ -742,6 +892,34 @@ aws iam attach-user-policy \ --user-name OpenChoreoCloudWatchLogsUser \ --policy-arn "arn:aws:iam::${AWS_ACCOUNT_ID}:policy/OpenChoreoCloudWatchLogsAdapterPolicy" +# Create and attach the setup job policy +aws iam create-policy \ + --policy-name OpenChoreoCloudWatchLogsSetupPolicy \ + --policy-document "$(cat </application`. | | `amazon-cloudwatch-observability.clusterName` | `openchoreo` | Upstream Container Insights chart value required to render the dependency. This module does not use it for log group naming. | | `amazon-cloudwatch-observability.region` | Required | AWS region for CloudWatch log groups and API calls. | | `awsCredentials.create` | `false` | Creates a static AWS credentials Secret. Keep `false` for Pod Identity, IRSA, or instance-profile based auth. Set to `true` for k3d, kind, or non-EKS clusters. | | `awsCredentials.name` | `""` | Name of the AWS credentials Secret. Required when `awsCredentials.create=true`. | | `awsCredentials.accessKeyId` | Required if `create=true` | AWS access key ID. | | `awsCredentials.secretAccessKey` | Required if `create=true` | AWS secret access key. | -| `containerLogs.retentionDays` | `7` | Retention period applied to log groups created by the setup Job. Must be one of the retention values supported by CloudWatch Logs. | +| `containerLogs.retentionDays` | `7` | Retention period applied to the application log group created by the setup Job. Must be one of the retention values supported by CloudWatch Logs. | +| `events.provisionLogGroup` | `true` | Provisions the `/events` log group (and retention) for Kubernetes events ingested by `observability-events-otel-collector`. Set to `false` only when the install intentionally does not grant events log group IAM permissions. The adapter serves `/api/v1/events/query` regardless and returns an empty result when the group is absent. | +| `events.logGroupName` | `""` | Optional exact events log group name. When empty, the chart uses `/events`. Set the events collector `awscloudwatchlogs.log_group_name` to the same value. | +| `events.retentionDays` | `30` | Retention period applied to the events log group. Must be one of the retention values supported by CloudWatch Logs. | | `cloudWatchAgent.enabled` | `true` | Enables the upstream `amazon-cloudwatch-observability` subchart. Set to `false` on an observability-plane-only install. | | `cloudWatchAgent.bridgeService.enabled` | `false` | Optional compatibility bridge from `amazon-cloudwatch/cloudwatch-agent` to the real Service. Only needed if Pod Association is re-enabled in Fluent Bit. | | `cloudWatchAgent.injectAwsCredentials.enabled` | `false` | Patches Fluent Bit to consume static AWS credentials. Enable this with `awsCredentials.create=true` for non-EKS clusters. | diff --git a/observability-logs-aws-cloudwatch/VERSION b/observability-logs-aws-cloudwatch/VERSION index 0c62199f..0d91a54c 100644 --- a/observability-logs-aws-cloudwatch/VERSION +++ b/observability-logs-aws-cloudwatch/VERSION @@ -1 +1 @@ -0.2.1 +0.3.0 diff --git a/observability-logs-aws-cloudwatch/helm/templates/_helpers.tpl b/observability-logs-aws-cloudwatch/helm/templates/_helpers.tpl index 6f24729b..002a0511 100644 --- a/observability-logs-aws-cloudwatch/helm/templates/_helpers.tpl +++ b/observability-logs-aws-cloudwatch/helm/templates/_helpers.tpl @@ -21,8 +21,24 @@ AWS region sourced from the upstream subchart values block. Resolved application log group name for the adapter and setup Job. */}} {{- define "logs-aws-cloudwatch.logGroupName" -}} +{{- if .Values.global.applicationLogGroupName -}} +{{- .Values.global.applicationLogGroupName -}} +{{- else -}} {{- printf "%s/application" (include "logs-aws-cloudwatch.logGroupPrefix" .) -}} {{- end -}} +{{- end -}} + +{{/* +Log group where the observability-events-otel-collector ships enriched Kubernetes +events. Mirrors the application log group under the shared prefix. +*/}} +{{- define "logs-aws-cloudwatch.eventsLogGroupName" -}} +{{- if .Values.events.logGroupName -}} +{{- .Values.events.logGroupName -}} +{{- else -}} +{{- printf "%s/events" (include "logs-aws-cloudwatch.logGroupPrefix" .) -}} +{{- end -}} +{{- end -}} {{/* Validate required values and fail fast with a readable message. diff --git a/observability-logs-aws-cloudwatch/helm/templates/adapter/configmap.yaml b/observability-logs-aws-cloudwatch/helm/templates/adapter/configmap.yaml index 334170cd..ffc8af47 100644 --- a/observability-logs-aws-cloudwatch/helm/templates/adapter/configmap.yaml +++ b/observability-logs-aws-cloudwatch/helm/templates/adapter/configmap.yaml @@ -11,6 +11,7 @@ data: AWS_REGION: {{ include "logs-aws-cloudwatch.region" . | quote }} LOG_GROUP_PREFIX: {{ include "logs-aws-cloudwatch.logGroupPrefix" . | quote }} LOG_GROUP_NAME: {{ include "logs-aws-cloudwatch.logGroupName" . | quote }} + EVENTS_LOG_GROUP_NAME: {{ include "logs-aws-cloudwatch.eventsLogGroupName" . | quote }} QUERY_TIMEOUT_SECONDS: {{ .Values.adapter.queryTimeoutSeconds | quote }} QUERY_POLL_MILLISECONDS: {{ .Values.adapter.queryPollMilliseconds | quote }} LOG_LEVEL: {{ .Values.adapter.logLevel | quote }} diff --git a/observability-logs-aws-cloudwatch/helm/templates/cloudwatch-setup/job.yaml b/observability-logs-aws-cloudwatch/helm/templates/cloudwatch-setup/job.yaml index 42d261a5..d7f918dc 100644 --- a/observability-logs-aws-cloudwatch/helm/templates/cloudwatch-setup/job.yaml +++ b/observability-logs-aws-cloudwatch/helm/templates/cloudwatch-setup/job.yaml @@ -30,8 +30,16 @@ spec: value: {{ include "logs-aws-cloudwatch.region" . | quote }} - name: LOG_GROUP_PREFIX value: {{ include "logs-aws-cloudwatch.logGroupPrefix" . | quote }} + - name: LOG_GROUP_NAME + value: {{ include "logs-aws-cloudwatch.logGroupName" . | quote }} - name: RETENTION_DAYS value: {{ .Values.containerLogs.retentionDays | quote }} + - name: EVENTS_PROVISION_LOG_GROUP + value: {{ .Values.events.provisionLogGroup | quote }} + - name: EVENTS_LOG_GROUP_NAME + value: {{ include "logs-aws-cloudwatch.eventsLogGroupName" . | quote }} + - name: EVENTS_RETENTION_DAYS + value: {{ .Values.events.retentionDays | quote }} {{- if .Values.awsCredentials.name }} envFrom: - secretRef: diff --git a/observability-logs-aws-cloudwatch/helm/values.yaml b/observability-logs-aws-cloudwatch/helm/values.yaml index 794659d9..dd055319 100644 --- a/observability-logs-aws-cloudwatch/helm/values.yaml +++ b/observability-logs-aws-cloudwatch/helm/values.yaml @@ -4,6 +4,9 @@ # Job. The application log group resolves to `/application`. global: logGroupPrefix: "/aws/containerinsights" + # Optional exact application log group name. When empty, the chart uses + # `/application`. + applicationLogGroupName: "" # AWS authentication mode. Defaults to **EKS Pod Identity / IRSA / EC2 # instance-profile** — the AWS SDK default credentials chain picks them up @@ -26,6 +29,20 @@ awsCredentials: containerLogs: retentionDays: 7 +# Kubernetes event support. Events are an optional, opt-in capability: they are +# ingested by the separate `observability-events-otel-collector` chart (deployed +# with its `awscloudwatchlogs` exporter pointed at the events log group below). +# The adapter always serves `POST /api/v1/events/query`. This flag only governs +# whether the setup Job provisions the events log group + retention. +events: + provisionLogGroup: true + # Optional exact events log group name. When empty, the chart uses + # `/events`. + logGroupName: "" + # Retention for the events log group. Sparser and longer-lived than container + # logs; must be one of CloudWatch's allowed values (see containerLogs above). + retentionDays: 30 + # --------------------------------------------------------------------------- # amazon-cloudwatch-observability dependency # --------------------------------------------------------------------------- @@ -148,7 +165,7 @@ amazon-cloudwatch-observability: Name cloudwatch_logs Match application.* region ${AWS_REGION} - log_group_name {{ .Values.global.logGroupPrefix | default "/aws/containerinsights" | trimSuffix "/" }}/application + log_group_name {{ if .Values.global.applicationLogGroupName }}{{ .Values.global.applicationLogGroupName }}{{ else }}{{ .Values.global.logGroupPrefix | default "/aws/containerinsights" | trimSuffix "/" }}/application{{ end }} log_stream_prefix ${HOST_NAME}- auto_create_group true extra_user_agent container-insights @@ -216,7 +233,7 @@ amazon-cloudwatch-observability: Name cloudwatch_logs Match application.* region ${AWS_REGION} - log_group_name {{ .Values.global.logGroupPrefix | default "/aws/containerinsights" | trimSuffix "/" }}/application + log_group_name {{ if .Values.global.applicationLogGroupName }}{{ .Values.global.applicationLogGroupName }}{{ else }}{{ .Values.global.logGroupPrefix | default "/aws/containerinsights" | trimSuffix "/" }}/application{{ end }} log_stream_prefix ${HOST_NAME}- auto_create_group true extra_user_agent container-insights diff --git a/observability-logs-aws-cloudwatch/init/setup-cloudwatch.sh b/observability-logs-aws-cloudwatch/init/setup-cloudwatch.sh index f5a2294c..878211be 100644 --- a/observability-logs-aws-cloudwatch/init/setup-cloudwatch.sh +++ b/observability-logs-aws-cloudwatch/init/setup-cloudwatch.sh @@ -10,31 +10,52 @@ set -euo pipefail AWS_REGION="${AWS_REGION:?AWS_REGION is required}" LOG_GROUP_PREFIX="${LOG_GROUP_PREFIX:-/aws/containerinsights}" +LOG_GROUP_NAME="${LOG_GROUP_NAME:-}" RETENTION_DAYS="${RETENTION_DAYS:-7}" +EVENTS_PROVISION_LOG_GROUP="${EVENTS_PROVISION_LOG_GROUP:-${EVENTS_ENABLED:-false}}" +EVENTS_LOG_GROUP_NAME="${EVENTS_LOG_GROUP_NAME:-}" +EVENTS_RETENTION_DAYS="${EVENTS_RETENTION_DAYS:-30}" # CloudWatch's PutRetentionPolicy only accepts this fixed set of values; any # other number is rejected by the API mid-run, leaving log groups created but # unconfigured. Fail fast with a clearer message instead. -case " 1 3 5 7 14 30 60 90 120 150 180 365 400 545 731 1096 1827 2192 2557 2922 3288 3653 " in - *" ${RETENTION_DAYS} "*) ;; - *) - echo "RETENTION_DAYS=${RETENTION_DAYS} is not a valid CloudWatch retention." >&2 - echo "Allowed: 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1096, 1827, 2192, 2557, 2922, 3288, 3653" >&2 - exit 1 - ;; -esac +validate_retention() { + case " 1 3 5 7 14 30 60 90 120 150 180 365 400 545 731 1096 1827 2192 2557 2922 3288 3653 " in + *" ${1} "*) ;; + *) + echo "Retention '${1}' is not a valid CloudWatch retention." >&2 + echo "Allowed: 1, 3, 5, 7, 14, 30, 60, 90, 120, 150, 180, 365, 400, 545, 731, 1096, 1827, 2192, 2557, 2922, 3288, 3653" >&2 + exit 1 + ;; + esac +} + +validate_retention "${RETENTION_DAYS}" # Strip any trailing slash so joining the segments stays predictable. LOG_GROUP_PREFIX="${LOG_GROUP_PREFIX%/}" +LOG_GROUP_NAME="${LOG_GROUP_NAME:-${LOG_GROUP_PREFIX}/application}" +EVENTS_LOG_GROUP_NAME="${EVENTS_LOG_GROUP_NAME:-${LOG_GROUP_PREFIX}/events}" +# Each entry is " ". The application log group is +# always provisioned; the events log group is opt-in +# (events.provisionLogGroup) and gets its own, typically longer, retention. LOG_GROUPS=( - "${LOG_GROUP_PREFIX}/application" + "${LOG_GROUP_NAME} ${RETENTION_DAYS}" ) -echo "Ensuring CloudWatch log groups exist in region ${AWS_REGION} with ${RETENTION_DAYS}-day retention" +if [ "${EVENTS_PROVISION_LOG_GROUP}" = "true" ]; then + validate_retention "${EVENTS_RETENTION_DAYS}" + LOG_GROUPS+=("${EVENTS_LOG_GROUP_NAME} ${EVENTS_RETENTION_DAYS}") +fi + +echo "Ensuring CloudWatch log groups exist in region ${AWS_REGION}" -for group in "${LOG_GROUPS[@]}"; do - echo "Ensuring log group: ${group}" +for entry in "${LOG_GROUPS[@]}"; do + # Split " " into its two fields. + group="${entry%% *}" + retention="${entry##* }" + echo "Ensuring log group: ${group} (retention ${retention}d)" # Attempt creation directly and key off the API's own "already exists" error. # This avoids a describe→create TOCTOU race and stops a real failure @@ -55,11 +76,11 @@ for group in "${LOG_GROUPS[@]}"; do exit "${create_status}" fi - echo "Setting retention on ${group} to ${RETENTION_DAYS} days" + echo "Setting retention on ${group} to ${retention} days" aws logs put-retention-policy \ --region "${AWS_REGION}" \ --log-group-name "${group}" \ - --retention-in-days "${RETENTION_DAYS}" + --retention-in-days "${retention}" done echo "CloudWatch log groups are ready" diff --git a/observability-logs-aws-cloudwatch/internal/api/gen/models.gen.go b/observability-logs-aws-cloudwatch/internal/api/gen/models.gen.go index 26c81de3..1fa13f22 100644 --- a/observability-logs-aws-cloudwatch/internal/api/gen/models.gen.go +++ b/observability-logs-aws-cloudwatch/internal/api/gen/models.gen.go @@ -67,6 +67,12 @@ const ( Unauthorized ErrorResponseTitle = "unauthorized" ) +// Defines values for EventsQueryRequestSortOrder. +const ( + EventsQueryRequestSortOrderAsc EventsQueryRequestSortOrder = "asc" + EventsQueryRequestSortOrderDesc EventsQueryRequestSortOrder = "desc" +) + // Defines values for LogsQueryRequestLogLevels. const ( DEBUG LogsQueryRequestLogLevels = "DEBUG" @@ -77,8 +83,8 @@ const ( // Defines values for LogsQueryRequestSortOrder. const ( - Asc LogsQueryRequestSortOrder = "asc" - Desc LogsQueryRequestSortOrder = "desc" + LogsQueryRequestSortOrderAsc LogsQueryRequestSortOrder = "asc" + LogsQueryRequestSortOrderDesc LogsQueryRequestSortOrder = "desc" ) // AlertRuleRequest defines model for AlertRuleRequest. @@ -277,6 +283,90 @@ type ErrorResponse struct { // ErrorResponseTitle The error message type ErrorResponseTitle string +// EventEntry defines model for EventEntry. +type EventEntry struct { + // Message The event message + Message *string `json:"message,omitempty"` + + // Metadata The metadata of the event + Metadata *struct { + // ComponentName The OpenChoreo component name the event is associated with + ComponentName *string `json:"componentName,omitempty"` + + // ComponentUid The OpenChoreo component UID the event is associated with + ComponentUid *openapi_types.UUID `json:"componentUid,omitempty"` + + // EnvironmentName The OpenChoreo environment name the event is associated with + EnvironmentName *string `json:"environmentName,omitempty"` + + // EnvironmentUid The OpenChoreo environment UID the event is associated with + EnvironmentUid *openapi_types.UUID `json:"environmentUid,omitempty"` + + // NamespaceName The OpenChoreo namespace name the event is associated with + NamespaceName *string `json:"namespaceName,omitempty"` + + // ObjectKind The kind of the Kubernetes object the event involves (e.g. CronJob) + ObjectKind *string `json:"objectKind,omitempty"` + + // ObjectName The name of the Kubernetes object the event involves + ObjectName *string `json:"objectName,omitempty"` + + // ObjectNamespace The namespace of the Kubernetes object the event involves + ObjectNamespace *string `json:"objectNamespace,omitempty"` + + // ProjectName The OpenChoreo project name the event is associated with + ProjectName *string `json:"projectName,omitempty"` + + // ProjectUid The OpenChoreo project UID the event is associated with + ProjectUid *openapi_types.UUID `json:"projectUid,omitempty"` + } `json:"metadata,omitempty"` + + // Reason The short, machine-readable reason for the event (e.g. SawCompletedJob) + Reason *string `json:"reason,omitempty"` + + // Timestamp The timestamp of the event + Timestamp *time.Time `json:"timestamp,omitempty"` + + // Type The event type (e.g. Normal, Warning) + Type *string `json:"type,omitempty"` +} + +// EventsQueryRequest defines model for EventsQueryRequest. +type EventsQueryRequest struct { + // EndTime The end time of the query + EndTime time.Time `json:"endTime"` + + // Limit The maximum number of items to return + Limit *int `json:"limit,omitempty"` + SearchScope EventsQueryRequest_SearchScope `json:"searchScope"` + + // SortOrder The sort order of the query + SortOrder *EventsQueryRequestSortOrder `json:"sortOrder,omitempty"` + + // StartTime The start time of the query + StartTime time.Time `json:"startTime"` +} + +// EventsQueryRequest_SearchScope defines model for EventsQueryRequest.SearchScope. +type EventsQueryRequest_SearchScope struct { + union json.RawMessage +} + +// EventsQueryRequestSortOrder The sort order of the query +type EventsQueryRequestSortOrder string + +// EventsQueryResponse defines model for EventsQueryResponse. +type EventsQueryResponse struct { + // Events The events queried successfully + Events *[]EventEntry `json:"events,omitempty"` + + // TookMs The time taken to query the events in milliseconds + TookMs *int `json:"tookMs,omitempty"` + + // Total The total number of matching events, capped at 1000 + Total *int `json:"total,omitempty"` +} + // LogsQueryRequest defines model for LogsQueryRequest. type LogsQueryRequest struct { // EndTime The end time of the query @@ -340,13 +430,19 @@ type WorkflowLogEntry struct { // WorkflowSearchScope defines model for WorkflowSearchScope. type WorkflowSearchScope struct { - Namespace string `json:"namespace"` + Namespace string `json:"namespace"` + + // TaskName Filter events to a specific workflow task + TaskName *string `json:"taskName,omitempty"` WorkflowRunName *string `json:"workflowRunName,omitempty"` } // HandleAlertWebhookJSONBody defines parameters for HandleAlertWebhook. type HandleAlertWebhookJSONBody = map[string]interface{} +// QueryEventsJSONRequestBody defines body for QueryEvents for application/json ContentType. +type QueryEventsJSONRequestBody = EventsQueryRequest + // QueryLogsJSONRequestBody defines body for QueryLogs for application/json ContentType. type QueryLogsJSONRequestBody = LogsQueryRequest @@ -359,6 +455,68 @@ type UpdateAlertRuleJSONRequestBody = AlertRuleRequest // HandleAlertWebhookJSONRequestBody defines body for HandleAlertWebhook for application/json ContentType. type HandleAlertWebhookJSONRequestBody = HandleAlertWebhookJSONBody +// AsComponentSearchScope returns the union data inside the EventsQueryRequest_SearchScope as a ComponentSearchScope +func (t EventsQueryRequest_SearchScope) AsComponentSearchScope() (ComponentSearchScope, error) { + var body ComponentSearchScope + err := json.Unmarshal(t.union, &body) + return body, err +} + +// FromComponentSearchScope overwrites any union data inside the EventsQueryRequest_SearchScope as the provided ComponentSearchScope +func (t *EventsQueryRequest_SearchScope) FromComponentSearchScope(v ComponentSearchScope) error { + b, err := json.Marshal(v) + t.union = b + return err +} + +// MergeComponentSearchScope performs a merge with any union data inside the EventsQueryRequest_SearchScope, using the provided ComponentSearchScope +func (t *EventsQueryRequest_SearchScope) MergeComponentSearchScope(v ComponentSearchScope) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + + merged, err := runtime.JSONMerge(t.union, b) + t.union = merged + return err +} + +// AsWorkflowSearchScope returns the union data inside the EventsQueryRequest_SearchScope as a WorkflowSearchScope +func (t EventsQueryRequest_SearchScope) AsWorkflowSearchScope() (WorkflowSearchScope, error) { + var body WorkflowSearchScope + err := json.Unmarshal(t.union, &body) + return body, err +} + +// FromWorkflowSearchScope overwrites any union data inside the EventsQueryRequest_SearchScope as the provided WorkflowSearchScope +func (t *EventsQueryRequest_SearchScope) FromWorkflowSearchScope(v WorkflowSearchScope) error { + b, err := json.Marshal(v) + t.union = b + return err +} + +// MergeWorkflowSearchScope performs a merge with any union data inside the EventsQueryRequest_SearchScope, using the provided WorkflowSearchScope +func (t *EventsQueryRequest_SearchScope) MergeWorkflowSearchScope(v WorkflowSearchScope) error { + b, err := json.Marshal(v) + if err != nil { + return err + } + + merged, err := runtime.JSONMerge(t.union, b) + t.union = merged + return err +} + +func (t EventsQueryRequest_SearchScope) MarshalJSON() ([]byte, error) { + b, err := t.union.MarshalJSON() + return b, err +} + +func (t *EventsQueryRequest_SearchScope) UnmarshalJSON(b []byte) error { + err := t.union.UnmarshalJSON(b) + return err +} + // AsComponentSearchScope returns the union data inside the LogsQueryRequest_SearchScope as a ComponentSearchScope func (t LogsQueryRequest_SearchScope) AsComponentSearchScope() (ComponentSearchScope, error) { var body ComponentSearchScope diff --git a/observability-logs-aws-cloudwatch/internal/api/gen/server.gen.go b/observability-logs-aws-cloudwatch/internal/api/gen/server.gen.go index 08e5efa3..7ff3f139 100644 --- a/observability-logs-aws-cloudwatch/internal/api/gen/server.gen.go +++ b/observability-logs-aws-cloudwatch/internal/api/gen/server.gen.go @@ -24,6 +24,9 @@ import ( // ServerInterface represents all server handlers. type ServerInterface interface { + // Query events + // (POST /api/v1/events/query) + QueryEvents(w http.ResponseWriter, r *http.Request) // Query logs // (POST /api/v1/logs/query) QueryLogs(w http.ResponseWriter, r *http.Request) @@ -56,6 +59,20 @@ type ServerInterfaceWrapper struct { type MiddlewareFunc func(http.Handler) http.Handler +// QueryEvents operation middleware +func (siw *ServerInterfaceWrapper) QueryEvents(w http.ResponseWriter, r *http.Request) { + + handler := http.Handler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + siw.Handler.QueryEvents(w, r) + })) + + for _, middleware := range siw.HandlerMiddlewares { + handler = middleware(handler) + } + + handler.ServeHTTP(w, r) +} + // QueryLogs operation middleware func (siw *ServerInterfaceWrapper) QueryLogs(w http.ResponseWriter, r *http.Request) { @@ -307,6 +324,7 @@ func HandlerWithOptions(si ServerInterface, options StdHTTPServerOptions) http.H ErrorHandlerFunc: options.ErrorHandlerFunc, } + m.HandleFunc("POST "+options.BaseURL+"/api/v1/events/query", wrapper.QueryEvents) m.HandleFunc("POST "+options.BaseURL+"/api/v1/logs/query", wrapper.QueryLogs) m.HandleFunc("POST "+options.BaseURL+"/api/v1alpha1/alerts/rules", wrapper.CreateAlertRule) m.HandleFunc("DELETE "+options.BaseURL+"/api/v1alpha1/alerts/rules/{ruleName}", wrapper.DeleteAlertRule) @@ -318,6 +336,59 @@ func HandlerWithOptions(si ServerInterface, options StdHTTPServerOptions) http.H return m } +type QueryEventsRequestObject struct { + Body *QueryEventsJSONRequestBody +} + +type QueryEventsResponseObject interface { + VisitQueryEventsResponse(w http.ResponseWriter) error +} + +type QueryEvents200JSONResponse EventsQueryResponse + +func (response QueryEvents200JSONResponse) VisitQueryEventsResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(200) + + return json.NewEncoder(w).Encode(response) +} + +type QueryEvents400JSONResponse ErrorResponse + +func (response QueryEvents400JSONResponse) VisitQueryEventsResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(400) + + return json.NewEncoder(w).Encode(response) +} + +type QueryEvents401JSONResponse ErrorResponse + +func (response QueryEvents401JSONResponse) VisitQueryEventsResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(401) + + return json.NewEncoder(w).Encode(response) +} + +type QueryEvents403JSONResponse ErrorResponse + +func (response QueryEvents403JSONResponse) VisitQueryEventsResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(403) + + return json.NewEncoder(w).Encode(response) +} + +type QueryEvents500JSONResponse ErrorResponse + +func (response QueryEvents500JSONResponse) VisitQueryEventsResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(500) + + return json.NewEncoder(w).Encode(response) +} + type QueryLogsRequestObject struct { Body *QueryLogsJSONRequestBody } @@ -530,6 +601,15 @@ func (response UpdateAlertRule400JSONResponse) VisitUpdateAlertRuleResponse(w ht return json.NewEncoder(w).Encode(response) } +type UpdateAlertRule404JSONResponse ErrorResponse + +func (response UpdateAlertRule404JSONResponse) VisitUpdateAlertRuleResponse(w http.ResponseWriter) error { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(404) + + return json.NewEncoder(w).Encode(response) +} + type UpdateAlertRule500JSONResponse ErrorResponse func (response UpdateAlertRule500JSONResponse) VisitUpdateAlertRuleResponse(w http.ResponseWriter) error { @@ -606,6 +686,9 @@ func (response Health503JSONResponse) VisitHealthResponse(w http.ResponseWriter) // StrictServerInterface represents all server handlers. type StrictServerInterface interface { + // Query events + // (POST /api/v1/events/query) + QueryEvents(ctx context.Context, request QueryEventsRequestObject) (QueryEventsResponseObject, error) // Query logs // (POST /api/v1/logs/query) QueryLogs(ctx context.Context, request QueryLogsRequestObject) (QueryLogsResponseObject, error) @@ -658,6 +741,37 @@ type strictHandler struct { options StrictHTTPServerOptions } +// QueryEvents operation middleware +func (sh *strictHandler) QueryEvents(w http.ResponseWriter, r *http.Request) { + var request QueryEventsRequestObject + + var body QueryEventsJSONRequestBody + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + sh.options.RequestErrorHandlerFunc(w, r, fmt.Errorf("can't decode JSON body: %w", err)) + return + } + request.Body = &body + + handler := func(ctx context.Context, w http.ResponseWriter, r *http.Request, request interface{}) (interface{}, error) { + return sh.ssi.QueryEvents(ctx, request.(QueryEventsRequestObject)) + } + for _, middleware := range sh.middlewares { + handler = middleware(handler, "QueryEvents") + } + + response, err := handler(r.Context(), w, r, request) + + if err != nil { + sh.options.ResponseErrorHandlerFunc(w, r, err) + } else if validResponse, ok := response.(QueryEventsResponseObject); ok { + if err := validResponse.VisitQueryEventsResponse(w); err != nil { + sh.options.ResponseErrorHandlerFunc(w, r, err) + } + } else if response != nil { + sh.options.ResponseErrorHandlerFunc(w, r, fmt.Errorf("unexpected response type: %T", response)) + } +} + // QueryLogs operation middleware func (sh *strictHandler) QueryLogs(w http.ResponseWriter, r *http.Request) { var request QueryLogsRequestObject @@ -863,48 +977,54 @@ func (sh *strictHandler) Health(w http.ResponseWriter, r *http.Request) { // Base64 encoded, gzipped, json marshaled Swagger object var swaggerSpec = []string{ - "H4sIAAAAAAAC/+xa2XPbuBn/VzBoZ7adoSU5SR9Wb47jZN1646ydNA+JpwMRn0isQYAGQDuKx/97Bwcv", - "CdQV5+g0L4lMgN99/PAR9ziVRSkFCKPx9B7rNIeCuJ9HHJS5qDhcwE0F2thnpZIlKMPA7UiloMwwKVaX", - "QJAZB2p/UtCpYqXfh9/nYHJQyOSAiOWAVMUBMY3qVxJsFiXgKZ5JyYEI/JBgJgyoW8JX6b3NAdWrSM6R", - "YQUgI9FNBWqB5nKZU0teG8VEZqlbwYmRKk69XrVUKw1xmiCqAk8/4MzgBGfGPuLG/eNWb3CCBdzgqwh3", - "kyvQueQ0zr5ZRreEV7BWikBbVMUMlKV9xwSVd3HCfm0/mz0kWMFNxZR18Qfcui4w7HisY96urq0l5OxP", - "SI2VtgBDKDEkFmkhSN+xATOdlyCOc6lAomYzenf6otELJ3guVUEMnuKqYjQWCCBumZKi2JJRZ/vOrAQp", - "IM7ArjivbIxbu1OXJF1DyC2vUkPHFzGCpZLWF9voHrbuqPdS3DgjdPXoibDij6QfB7EQ0rJS3h79APLy", - "RZXyUW8kgk+QVsanFpcZmhEN1BtNb1TFM1gVaWlbE+KNqEmnjMY06pRhXUqh4Wcd/lmHO0H4s4r+P1bR", - "rQtfAUaxNC6IX+tHXHi2VPvqtErL6j+VJpk1ZAGFVIvwZyyhvkrNXS2v8Yr5Hma5lNfDRbMA7SQfsIxb", - "7Lv8zpOMuVwbYiodp+XXhkjVltVVmoJ2tlZKqohBB1VlIrP94XIh0mF1SVo3iFUJ/Roy5BoEsj+Gqmqq", - "gBjXGqqS1r9EmhORud8UONinsWjgRBsrItAjM1BhWQHakKKsbWVfQXoh0pjJrWjPSXoNgp4OJNrML6PT", - "F+hvNsPmShZIzrRtUjPGmVnUW/6+Xamwz89kxlLCh3hyv+x42tKxJeVdA2jJMdoZ1lYOwnjUAbHoOa5L", - "/JnMToTxydoPGw63wAc1RX455m2ZDb9Vp17kvW7zihYst9oEiMwQOMGToVb3erBDRHudaxsmJwZlICze", - "AFpzion7JR11iMnGppdKYQgToIZ1a7bsqlCnGW9luW7z3p/VXjBhb/s1HX0rDdv+v6N+paTDDP5VzUAJ", - "MKBRKemepHdBLUsMd+DlIcpWtqrhzK7q7AmY9oyAWClses+2ralTeRp2tiUe2K14t/J7CUSl+WUqS9gM", - "2rfIo/UQd4P5N5/YPaXYWfXEQpdhDOKQzbGkA4HkllEqKXQ7NShk/2PuqAyfSFFyy/T8+eXBvw8Pzg6e", - "PIn3kQF091tVEHGggFB7yg0824bUMvidac1Ehmrt0ZwBpxr9og1R5i0r4BdEBEW/gKDur5gYhhm+VtsO", - "59DKZ4TWA1cLrkhlcqnYZ9/dpZoxSkHYY6w0L2Ul/FREzDlLTT12E4RfOsud7AAlz2Sm/7CwfHDcGxQd", - "UEdQf34NCbJylFmTHAnmrGABFc5JxQ2eHk4mSQwBkE+sqArkj9aWGTNQaHuUUGAqZS0T9jgakwQXTIQ/", - "G8bWSpk/mXOZnVkQ4zR0tLyq3hkvTp6/e4UTfPr65TlO8Puji9c4wScXF+cX8RmCf0CUIgvnPnZTwamn", - "alQFFum5ZH+TK6Lj6aj71UAKOJ/j6Yd7/FcFczzFfxm3I/txmNePo7XkIVn/0nuprudc3vXeuXKHSWXO", - "FQXVc4lzB455xe5H0r6w7P7akKR5M2q3JqcGEbAye4fXUglreSVNSPetfrU+PYbKG5eZHgS92onMgKJw", - "0ptXnFsNWv820beVoxvMvhx21utbkardP0zJxoKR8vp3PdwWw7GxGSKYWl8mUME4ZxpSKWjnNN/JPiPN", - "0DDRLXXyvCAmzW1BrlsvA52glJQlUEQMsske4RErdSuKx1y51/Hle8CIWBavKLQeAdwFEheVqHHe/jDg", - "wc2J5zJMpg1JTS2B7dstlnsLpLDcl+3ENDp6c4p0CSmbs5S44QSFOROgndUsVUVS4xEgsVbMbGQQSkoD", - "ChWVNojZPm5B0UdhpBtNZxYnojtmckelI8l5QBoj5OBmjTtSwrnjqF1/KyUTxraaj8KHugtziwEKIkjW", - "PZprj2CCi51w9TCiVPKWUaBo5nOlkLTiMPpoOxdnKYTSEsx1VJI0B/RkZGO7UhxPcW5Mqafj8d3d3Yi4", - "5ZFU2Ti8q8dnp8cnry9PDp6MJqPcFLyDQ/CKzvUcxJY3dBTsd/TmFCf4FpT2LjkcTUaTMKAXpGR4ip+O", - "JqOn9tRNTO4CbExKNr49HFubjJvJXyl1ZNjzR2u9xkwRsOcH/kyKU1q/ZMXEPhRBm+eSLuooA+EYkbLk", - "IWTGf2o/8PL1blM1XME/D/2gD+1bhfLvlH4ymXwN/qHBOAH6ljsbbCQPCX62lTQNwu3hcYw7mHk/7BtC", - "rINfbWPfSvf+uSGi96m4JZxRpFrKzyaHj6RtTdzC8aC4kdcOZNdK9XD446n1bonss8nTR9LpsnLVGH2s", - "JpOn6afFZ/cDUE40EhKVoJyq0jXuWwZ3Ph3lHDWDBTSXMkFvwoF7RlSCGvCBZuSzbfAnnaEMtTBalvZ3", - "x3LtkeXxzPayS/MfXxL17Sny1575OgrETlSPGdieOvLkUaBvGVRFQWwV7ZRLKxaxIPODqwT4ym4MlZfw", - "MieHY//pZOya0HAFPnaDfERE79uv2K4S+5eb79JfqR6vXD/aqh4fPi7/2KeViBePWiOGTyR7FuevWS5/", - "/Xb8O/YgXAGhCwSfmDZ6+4T9hvlVJ0PvA01IsyP/JXJDoo3v7X+2cj74XONgIkfZF+75nlnnX+5n3VeC", - "InuGfvj69wOG/rPvEvpCGjR3I7ofMerrYFwb9QnOINI+XoFZiuKh485KGL8C8+1iuHdtar2zFBjF4PZn", - "+P6PhK8LwQ2xWxJFCjCgtJuv7XBJiNkd9myL6ztGuK7weBmEdLHg8tTkKsFlFUmgd+76RLwTbMog/+6P", - "Cb++ew8K91J+uCT+4fKnjsC9QE99fWnwfPEbEZSDRkaxLAPVXOlq+wQJ/h0Mc0+ie5frCyJ9NfUDJTST", - "dIFSItDMpuMC/fPy/DXyo8wEEY0om89B2fPtssQaFWSBNAja2VSSBZeE6hGOzkW/cf4sX4IbzJ3gUJQ7", - "o/9Mn43ps1eAR/MrB8JNboWOQq3jHNJrR9BvXLoNtnxsGK3mkaf/haHW/5rQ3lZrv5d78RZbXdpcMfml", - "lx4xjWo6zutPv0BIf4+yJ6MsQfgPfVOUSiHAX3wMl+bWXstriVTisVRtKS0Fl/d0al3fCZvgyStHNDy8", - "Xx0Q11CW8PajRQtj3NzoIbkfbqH+Y4Yb6UXeD7G7SqErdOzFIP3D1cN/AwAA//+wJDwhdDYAAA==", + "H4sIAAAAAAAC/+xbXXPbttL+Kxi870zbGVqSk5yL6s5xnNata+fYzclF4jkDkSsSNQjQAChH9fi/n8EX", + "RUmgTMl26k59k8gEsV/YfXaxAG9xKspKcOBa4fEtVmkBJbE/DxhIfV4zOIfrGpQ2zyopKpCagn0jFTyj", + "mgq+PgScTBhk5mcGKpW0cu/hTwXoAiTSBSBiOCBZM0BUoTAlwXpeAR7jiRAMCMd3CaZcg5wRtk7v9wJQ", + "GEViijQtAWmBrmuQczQVq5wW5JWWlOeGuhGcaCHj1MOooVoriNMEXpd4/BnnGic41+YR0/YfO3qNE8zh", + "Gl9GuOtCgioEy+Lsm2E0I6yGjVJ42rwuJyAN7RvKM3ETJ+zGdrPZXYIlXNdUmiX+jBdL5xm2Vqxl3rau", + "C0uIyR+QaiNtCZpkRJOYp3kn/Ug7zHRWAT8shASBmpfRx+N3jV44wVMhS6LxGNc1zWKOAHxGpeBlT0at", + "17dmxUkJcQZmxK7KvX5r3lQVSTcQssPr1NDheYxgJYVZiz66+1e31HvFb6wR2nosibC2HsmyH8RcSIla", + "OnssO5CTL6qU83otEHyFtNYutJjI0YQoyJzR1L2qOAbrIq281rh4I2rSgtGYRi0YVpXgCl5w+AWHW074", + "gqL/RBTtDXwlaEnTuCBubNnj/LMV7AthlVb1f2tFcmPIEkoh5/7PWEA9Ceauw2scMT/BpBDiqhs0S1BW", + "8g7L2MHlJb9xJGNLrjTRtYrTcmNdpIJlVZ2moKytpRQyYtBOVSnPTX64mPO0W12ShgSxLqEbQ5pcAUfm", + "RxeqphKItqmhrrLwi6cF4bn9nQED8zTmDYwobUSE7EB3ICwtQWlSVsFWZgpSc57GTG5Ee0vSK+DZcUeg", + "TdwwOn6HvjcRNpWiRGKiTJKaUEb1PLzyQz+oMM9PRE5Twrp4MjdseRro6El5WwdaWRhlDWuQg1AWXYCY", + "9xwGiD8R+RHXLliX3YbBDFinpsgNx1Zb5N2zQuhF5rWTVxSw7GjjICJHYAVPulLdaWeGiOY6mzZ0QTTK", + "gZt6A7LAKSbuQzJqF5N7k14quCaUg+zWrXllW4VaybiX5drJe3dWO5UJO9uvyei9NFzk/y31q0TWzeDX", + "egKSgwaFKpHtSHqbqmWF4Ra8XInSy1ahnNlWnR0Lph09IAaFTe7pm5payNOwMylxz7yKt4PfCyAyLS5S", + "UcH9RXuPONpc4t5j/vt37I5SbK96ZEqX7hrEVjaHIutwJDuMUpFBO1ODROY/arfK8JWUFTNMz95e7P1n", + "f+9k79WreB7pqO5+rkvC9ySQzOxyPc9FQlow+I0qRXmOgvZoSoFlCn2nNJH6d1rCd4jwDH0HPLN/xcTQ", + "VLON2rY4+1Q+IVlouJriitS6EJL+6bK7kBOaZcDNNlbo96LmrivCp4ymOrTdOGEX1nJHW5SSRzPguqMK", + "2Fgrg5n4eDndknvsfB6kpAoRpURKLWLcUF08elbfyGqbvfOu+Xc7XR+chR+m7wNz8Xa6Omf/lfIOPa8o", + "zyL50k1rc+MzwWag0PcwyAfoUAr+i5j80M3ytFebog/LzTx2LAi24vaAemC71dq5KniIR8aQUQJRXVtn", + "VQipE1SStKAcFnnFzWl6eE4g5y4X5MZkfrtT7nKbbcuRAJp9SpHwoBvLzbgX9tQQZAn6RCSnPP8B988l", + "6t81yHnn4aFPmx1y8Mx1Q71+a42xjfoxWlLfY5iSmmk83h+NkljuIV9pWZfINWoNM6qhVEgLJEHX0uRZ", + "/46lMUpwSbn/s2Fscm7u+rxquZITHM6mePz5Fv+/hCke4/8bLo5bh/6sdRitA++SzZM+CXk1ZeJmac6l", + "bQRKfSYzkEsGsMrjmA3M+0iYCavGDhUJaWZGGztNPdTZvZB658VcKT8XvJLGgZatfnmfO3YWp7NwBt4R", + "FsoKTiFDvlc3rRkzelifMRM3rVervFpEDJGSuL+FuPpNdQe77801nVq9EIpyVFLGqIJU8KwF2S231EJ3", + "ndnYoVYAlEQbKMs9+QSlpKogQ0QjEwAR8jEAOBH5PzH8mchPYAbMati4RYijd0dvP/6EE3x8+v4MJ/jT", + "wfkpTvDR+fnZefxAqu0kZi9Ar2s4dlS1rKHBmw+FJCq+t3sBpOcISK3w6IIjJnLV2UHthKLF+vYCpfUG", + "8Do29SQVlr+b0uWOOGf1fSqUC30cCrtC3ZrisaXcqRf+V/SkYlG8ptDmdpIm6iq+O3hPmQYZEpcWiCBV", + "QUqnNEU3njEy02PWCC+c1zxQ371VdWfvMkyFvz2hSaqDYniMWzuL34GUhvuq+alCBx+OG/GJPUDLYEo5", + "KLsYhqokdldFNCJmcXLjcCQjlTFCWSuNqNkGmL30F66FvT6RS6LB7lUslZYkZ74bNkB28xN6YylhzHJU", + "Nm1WgjrbfuEugmz0EJ6hknCSt4+PlOuyec+xwoUDs0qKGc0gQxMXgqXIagaDLyYhMpqCRyxvroOKpAWg", + "VwMTMrVkeIwLrSs1Hg5vbm4GxA4PhMyHfq4anhwfHp1eHO29GowGhS5Zq1eG13QOZ3UGNdGBt9/Bh2Oc", + "4BlI5ZZkfzAajPwlEk4qisf49WA0eI0TXBFdWL8dkooOZ/tD54DD5ny6EipyJGkBur1N9n7bWC3Sn3R3", + "VKjgx1mg4IpP7HwTlH4rsnlwO7NpG99iUlXM+9DwD7/VdLjaq6pcrrTuluPAFwrSJxprh1ej0dNI4JOZ", + "FWHZmEcbKui7BL/pJVHTnF1qJWPcavfu1rb1ntdqvZoyop/+Sy3viObHfEYYzZBcUH4z2n8kbQNxIVHp", + "FdfiyvaHg1JLLeTHU+vjCtk3o9ePpNNF7dpRX+rR6HX6df6n/QGoIApxgSqQVlVhy4QZhZsQmGKKmiYY", + "mgqRoA++LzQhMkFNsYMm5E9TUBy1epmZKdtFVbo2SrDdot/+eIZ736b5r4f4vT8COdob7S8ZsKVA7Djg", + "MV3bUUeOPPL0DYO6LIkBV4+iEEBQE1PWfvZ4gC/NywGWTarqB8o2qW2JwyZ7PBEKr+12vzEGr28nIot1", + "0rlteMHfF/x9EP7acPyHou/J3qsfnzv6Mgd9AXstEraRl7CqIPtDd+tyaPcG3Qh8aO8AIsKXro3zfkjs", + "JjdX2p8Ij9e+XOqFx/uPyz92KzOyigcLI/rblTuC81PC5Y/fjn/LHoRJINkcwVeqtOofsN8wvkIwLN3t", + "9GF24C4x3xNow1vzn0HOOxdrDHSkdfLOPt8x6tzk5ah7olJkR9f3F4efoeu/+UtcnwuNpvZ2z3P0+uCM", + "G70+wTlE0sdPoFe8uKsLtebGP4H+dj689MXV5sWSoCWF2Yv7/k3c17rgPb5bEUlK0CCVPU3Z4vsiat6o", + "iL1s4hukAeHxahHSrgVXm9mXCa7qSAB9tF9exDPBfRHk5j7P8usvz0H+k5aXIP5bBHEIg50qr/D5Vecm", + "52fCMwYKaUnzHGTzSdoiWRHvZJ2x5ki0v0V7QLit44+nhCYim6OUcDQxmDBHv1ycnfoLfQkiCmV0OgVp", + "NtmrEitUkjlSwLPWSxWZM0EyNcDRM7NvHMSrH/F1OqxfUFRYoz+/GH524bOTg0fjqwDCdGGEjtZ7hwWk", + "V5age3Hla7bVvctgPY4c/Qe62vIB9uJru8V9fyfevNdHp2smv3DSI6pQoGNX/fUDhHTfgS7JKCrg7m7J", + "GKWCc3AfbvqP/jZ+VrggUvPHUnVBacW53EqnZulbbuNX8tIS9Q9v17vUoZ4mbHGgvailbPPqLrntzlvu", + "oNv2FSPzve+uU2gLHZvopb+7vPtfAAAA///9tcKSNEcAAA==", } // GetSwagger returns the content of the embedded swagger specification file diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/client.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/client.go index dbeb3a9d..23b8027e 100644 --- a/observability-logs-aws-cloudwatch/internal/cloudwatch/client.go +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/client.go @@ -64,6 +64,7 @@ type Client struct { alarms alarmsAPI sts stsAPI logGroupName string + eventsLogGroupName string queryTimeout time.Duration pollEvery time.Duration alertMetricNamespace string @@ -77,6 +78,7 @@ type Client struct { type Config struct { Region string LogGroupName string + EventsLogGroupName string QueryTimeout time.Duration PollEvery time.Duration AlertMetricNamespace string @@ -98,6 +100,7 @@ func NewClient(ctx context.Context, cfg Config, logger *slog.Logger) (*Client, e alarms: cloudwatch.NewFromConfig(awsCfg), sts: sts.NewFromConfig(awsCfg), logGroupName: cfg.LogGroupName, + eventsLogGroupName: cfg.EventsLogGroupName, queryTimeout: cfg.QueryTimeout, pollEvery: cfg.PollEvery, alertMetricNamespace: cfg.AlertMetricNamespace, @@ -116,6 +119,7 @@ func NewClientWithAWS(logs logsAPI, alarms alarmsAPI, stsClient stsAPI, cfg Conf alarms: alarms, sts: stsClient, logGroupName: cfg.LogGroupName, + eventsLogGroupName: cfg.EventsLogGroupName, queryTimeout: cfg.QueryTimeout, pollEvery: cfg.PollEvery, alertMetricNamespace: cfg.AlertMetricNamespace, diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/event_labels.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_labels.go new file mode 100644 index 00000000..8fd5d424 --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_labels.go @@ -0,0 +1,71 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import "fmt" + +// CloudWatch Logs Insights field paths for querying Kubernetes events. +// +// Events are NOT collected by Fluent Bit (which produces the `kubernetes.labels.*` +// shape the container-log query builder targets). They are collected by the +// observability-events-otel-collector distribution and shipped by the OTEL +// `awscloudwatchlogsexporter`, which serialises each log record to a JSON envelope +// of the form: +// +// { +// "body": "", +// "severity_text": "Normal" | "Warning", +// "attributes": { "k8s.event.reason": "...", "k8s.namespace.name": "...", ... }, +// "resource": { "k8s.object.kind": "...", "k8s.object.name": "...", +// "k8s.object.label.openchoreo.dev/component-uid": "...", ... } +// } +// +// CloudWatch Logs Insights auto-discovers JSON fields and flattens nested objects +// with dot notation, so the enriched OpenChoreo labels land at +// `resource.k8s.object.label.`. This is the CloudWatch analog of the +// OpenSearch adapter's event_labels.go. +// +// NOTE: These paths mirror the awscloudwatchlogsexporter's documented serialisation +// (severity_text/attributes/resource) and must be confirmed against real records in +// the events log group. The container-log `labelField` / `label*` constants MUST NOT +// be reused here — they target the Fluent Bit shape. +const ( + // evTimestamp is the built-in Insights ingestion timestamp column. + evTimestamp = "@timestamp" + // evMessage is the event message, stored as the OTEL log record body. + evMessage = "body" + // evSeverityText is the event type (Normal/Warning), stored as OTEL severity text. + evSeverityText = "severity_text" + // evReason is the short, machine-readable reason for the event. + evReason = "attributes.k8s.event.reason" + // evObjectNamespace is the Kubernetes namespace the event was emitted in. + evObjectNamespace = "attributes.k8s.namespace.name" + // evObjectKind is the kind of the Kubernetes object the event involves. + evObjectKind = "resource.k8s.object.kind" + // evObjectName is the name of the Kubernetes object the event involves. + evObjectName = "resource.k8s.object.name" + + // evLabelPrefix is the flattened path prefix for OpenChoreo labels copied onto + // the event's involved object by the k8seventenrich processor. + evLabelPrefix = "resource.k8s.object.label." +) + +// Event label field paths, derived from the same OpenChoreo label keys the +// container-log query builder uses (labelComponentUID, etc.). +var ( + evComponentUID = evLabelPrefix + labelComponentUID + evComponentName = evLabelPrefix + labelComponentName + evEnvironmentUID = evLabelPrefix + labelEnvironmentUID + evEnvironmentName = evLabelPrefix + labelEnvironmentName + evProjectUID = evLabelPrefix + labelProjectUID + evProjectName = evLabelPrefix + labelProjectName + evNamespaceName = evLabelPrefix + labelNamespace +) + +// eventField renders a CloudWatch Insights accessor for an event field path. +// Insights flattens nested JSON into dotted field names; those paths contain dots +// (and, for labels, slashes) so the whole path must be wrapped in backticks. +func eventField(path string) string { + return fmt.Sprintf("`%s`", path) +} diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries.go new file mode 100644 index 00000000..4e5a578c --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries.go @@ -0,0 +1,117 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import ( + "fmt" + "strings" +) + +// eventResultFields lists the aliased columns every events query projects. The +// aliases become the keys of the row maps runQuery returns, so the client mapping +// can read them by clean name instead of the raw flattened JSON path. +var eventResultFields = []struct { + source string + alias string +}{ + {evMessage, "message"}, + {evSeverityText, "type"}, + {evReason, "reason"}, + {evObjectKind, "objectKind"}, + {evObjectName, "objectName"}, + {evObjectNamespace, "objectNamespace"}, + {evComponentUID, "componentUid"}, + {evComponentName, "componentName"}, + {evEnvironmentUID, "environmentUid"}, + {evEnvironmentName, "environmentName"}, + {evProjectUID, "projectUid"}, + {evProjectName, "projectName"}, + {evNamespaceName, "namespaceName"}, +} + +// writeEventFields writes the shared `fields @timestamp, ` projection. +func writeEventFields(b *strings.Builder) { + b.WriteString("fields @timestamp") + for _, f := range eventResultFields { + fmt.Fprintf(b, ", %s as %s", eventField(f.source), f.alias) + } + b.WriteString("\n") +} + +// buildComponentEventsQuery constructs a CloudWatch Logs Insights query for +// component-scoped Kubernetes events. It filters on the OpenChoreo scope labels the +// k8seventenrich processor copied onto each event's involved object — mirroring +// buildComponentQuery, but against the OTEL/awscloudwatchlogsexporter field shape. +func buildComponentEventsQuery(p ComponentEventsParams) string { + var b strings.Builder + + writeEventFields(&b) + + fmt.Fprintf(&b, "| filter %s = \"%s\"\n", eventField(evNamespaceName), escapeInsights(p.Namespace)) + if p.ProjectID != "" { + fmt.Fprintf(&b, "| filter %s = \"%s\"\n", eventField(evProjectUID), escapeInsights(p.ProjectID)) + } + if p.EnvironmentID != "" { + fmt.Fprintf(&b, "| filter %s = \"%s\"\n", eventField(evEnvironmentUID), escapeInsights(p.EnvironmentID)) + } + if p.ComponentID != "" { + fmt.Fprintf(&b, "| filter %s = \"%s\"\n", eventField(evComponentUID), escapeInsights(p.ComponentID)) + } + + fmt.Fprintf(&b, "| sort @timestamp %s\n", normaliseSortOrder(p.SortOrder)) + fmt.Fprintf(&b, "| limit %d", insightsLimit(p.Limit)) + + return b.String() +} + +// buildWorkflowEventsQuery constructs a CloudWatch Logs Insights query for +// workflow-scoped events. Workflow pods/jobs are named "-<...>" and +// run in the "workflows-" Kubernetes namespace, so — like the workflow +// log query — events are matched by the involved object's name and namespace rather +// than the OpenChoreo component labels. +func buildWorkflowEventsQuery(p WorkflowEventsParams) string { + var b strings.Builder + + writeEventFields(&b) + + fmt.Fprintf(&b, "| filter %s = \"workflows-%s\"\n", eventField(evObjectNamespace), escapeInsights(p.Namespace)) + if p.WorkflowRunName != "" { + // Match the workflow object itself ("") and its pods/jobs ("-<...>"). + fmt.Fprintf(&b, "| filter %s like /^%s(-|$)/\n", eventField(evObjectName), escapeInsightsRegex(p.WorkflowRunName)) + } + if p.TaskName != "" { + // Workflow controller events keep the step name in the message body, while + // pod/job events may include either the step name or the referenced template + // name in the object name. For example, the checkout-source step creates + // pods named "-checkout-". + taskName := escapeInsights(p.TaskName) + fmt.Fprintf(&b, "| filter %s like \"%s\" or %s like \"%s\"", + eventField(evObjectName), taskName, + eventField(evMessage), taskName, + ) + if p.WorkflowRunName != "" { + for _, objectPrefix := range workflowTaskObjectPrefixes(p.TaskName) { + fmt.Fprintf(&b, " or %s like /^%s-%s(-|$)/", + eventField(evObjectName), + escapeInsightsRegex(p.WorkflowRunName), + escapeInsightsRegex(objectPrefix), + ) + } + } + b.WriteString("\n") + } + + fmt.Fprintf(&b, "| sort @timestamp %s\n", normaliseSortOrder(p.SortOrder)) + fmt.Fprintf(&b, "| limit %d", insightsLimit(p.Limit)) + + return b.String() +} + +func workflowTaskObjectPrefixes(taskName string) []string { + prefixes := []string{taskName} + if cut, _, ok := strings.Cut(taskName, "-"); ok && cut != "" && cut != taskName { + prefixes = append(prefixes, cut) + } + return prefixes +} diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries_test.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries_test.go new file mode 100644 index 00000000..a4c22e71 --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_queries_test.go @@ -0,0 +1,119 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import ( + "strings" + "testing" +) + +func TestBuildComponentEventsQueryProjectsEventColumns(t *testing.T) { + query := buildComponentEventsQuery(ComponentEventsParams{ + Namespace: "default", + }) + + // Event columns must use the OTEL/awscloudwatchlogsexporter field shape, not the + // Fluent Bit kubernetes.labels.* shape used by container logs. + wantColumns := []string{ + "`body` as message", + "`severity_text` as type", + "`attributes.k8s.event.reason` as reason", + "`resource.k8s.object.kind` as objectKind", + "`resource.k8s.object.name` as objectName", + "`attributes.k8s.namespace.name` as objectNamespace", + "`resource.k8s.object.label.openchoreo.dev/component-uid` as componentUid", + "`resource.k8s.object.label.openchoreo.dev/namespace` as namespaceName", + } + for _, want := range wantColumns { + if !strings.Contains(query, want) { + t.Errorf("query missing column %q\nquery:\n%s", want, query) + } + } + + if strings.Contains(query, "kubernetes.labels.") { + t.Errorf("events query must not reference the Fluent Bit kubernetes.labels.* shape\nquery:\n%s", query) + } +} + +func TestBuildComponentEventsQueryFiltersScope(t *testing.T) { + query := buildComponentEventsQuery(ComponentEventsParams{ + Namespace: "default", + ProjectID: "proj-1", + EnvironmentID: "env-1", + ComponentID: "comp-1", + SortOrder: "asc", + Limit: 50, + }) + + wantFilters := []string{ + "| filter `resource.k8s.object.label.openchoreo.dev/namespace` = \"default\"", + "| filter `resource.k8s.object.label.openchoreo.dev/project-uid` = \"proj-1\"", + "| filter `resource.k8s.object.label.openchoreo.dev/environment-uid` = \"env-1\"", + "| filter `resource.k8s.object.label.openchoreo.dev/component-uid` = \"comp-1\"", + "| sort @timestamp asc", + "| limit 50", + } + for _, want := range wantFilters { + if !strings.Contains(query, want) { + t.Errorf("query missing %q\nquery:\n%s", want, query) + } + } +} + +func TestBuildComponentEventsQueryOmitsUnsetScope(t *testing.T) { + query := buildComponentEventsQuery(ComponentEventsParams{Namespace: "default"}) + + for _, unwanted := range []string{"project-uid", "environment-uid", "component-uid"} { + if strings.Contains(query, "filter `resource.k8s.object.label.openchoreo.dev/"+unwanted+"`") { + t.Errorf("query should not filter on unset %s\nquery:\n%s", unwanted, query) + } + } + // Default sort order is descending. + if !strings.Contains(query, "| sort @timestamp desc") { + t.Errorf("expected default desc sort\nquery:\n%s", query) + } +} + +func TestBuildWorkflowEventsQueryMatchesObjectNameAndNamespace(t *testing.T) { + query := buildWorkflowEventsQuery(WorkflowEventsParams{ + Namespace: "default", + WorkflowRunName: "build-123", + TaskName: "clone-step", + }) + + wantFilters := []string{ + "| filter `attributes.k8s.namespace.name` = \"workflows-default\"", + "| filter `resource.k8s.object.name` like /^build-123(-|$)/", + "| filter `resource.k8s.object.name` like \"clone-step\" or `body` like \"clone-step\" or `resource.k8s.object.name` like /^build-123-clone-step(-|$)/ or `resource.k8s.object.name` like /^build-123-clone(-|$)/", + } + for _, want := range wantFilters { + if !strings.Contains(query, want) { + t.Errorf("query missing %q\nquery:\n%s", want, query) + } + } +} + +func TestBuildWorkflowEventsQueryMatchesCheckoutTemplateObjectName(t *testing.T) { + query := buildWorkflowEventsQuery(WorkflowEventsParams{ + Namespace: "default", + WorkflowRunName: "test-service-1783662896161", + TaskName: "checkout-source", + }) + + want := "`resource.k8s.object.name` like /^test-service-1783662896161-checkout(-|$)/" + if !strings.Contains(query, want) { + t.Errorf("query missing checkout template object-name filter %q\nquery:\n%s", want, query) + } +} + +func TestBuildWorkflowEventsQueryWithoutRunName(t *testing.T) { + query := buildWorkflowEventsQuery(WorkflowEventsParams{Namespace: "default"}) + + if !strings.Contains(query, "| filter `attributes.k8s.namespace.name` = \"workflows-default\"") { + t.Errorf("expected namespace filter\nquery:\n%s", query) + } + if strings.Contains(query, "`resource.k8s.object.name` like") { + t.Errorf("query should not add an object-name filter when run name is empty\nquery:\n%s", query) + } +} diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/event_types.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_types.go new file mode 100644 index 00000000..40e49429 --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/event_types.go @@ -0,0 +1,54 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import "time" + +// ComponentEventsParams captures the component-scoped event query options. +type ComponentEventsParams struct { + Namespace string + ProjectID string + EnvironmentID string + ComponentID string + StartTime time.Time + EndTime time.Time + Limit int + SortOrder string +} + +// WorkflowEventsParams captures the workflow-scoped event query options. +type WorkflowEventsParams struct { + Namespace string + WorkflowRunName string + TaskName string + StartTime time.Time + EndTime time.Time + Limit int + SortOrder string +} + +// EventEntry is one CloudWatch event record normalised for the Observer response. +type EventEntry struct { + Timestamp time.Time + Message string + Type string + Reason string + ObjectKind string + ObjectName string + ObjectNamespace string + ComponentUID string + ComponentName string + EnvironmentUID string + EnvironmentName string + ProjectUID string + ProjectName string + NamespaceName string +} + +// EventsResult wraps the entries and query metadata. +type EventsResult struct { + Events []EventEntry + TotalCount int + Took int +} diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/events.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/events.go new file mode 100644 index 00000000..cf936a08 --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/events.go @@ -0,0 +1,91 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import ( + "context" + "errors" + "log/slog" + "time" + + cwltypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types" +) + +// eventsLogGroup returns the CloudWatch log group where the events collector's +// awscloudwatchlogsexporter ships enriched Kubernetes events. +func (c *Client) eventsLogGroup() string { + return c.eventsLogGroupName +} + +// GetComponentEvents runs a Logs Insights query for component-scoped events. +func (c *Client) GetComponentEvents(ctx context.Context, params ComponentEventsParams) (*EventsResult, error) { + return c.runEventsQuery(ctx, buildComponentEventsQuery(params), params.StartTime, params.EndTime) +} + +// GetWorkflowEvents runs a Logs Insights query for workflow-scoped events. +func (c *Client) GetWorkflowEvents(ctx context.Context, params WorkflowEventsParams) (*EventsResult, error) { + return c.runEventsQuery(ctx, buildWorkflowEventsQuery(params), params.StartTime, params.EndTime) +} + +// runEventsQuery executes an events query and maps the rows to EventEntry values. +// +// The events collector is an optional, opt-in component: when it is not deployed the +// events log group does not exist. Rather than surfacing that as a 500, treat a +// missing log group as an empty result so /api/v1/events/query degrades gracefully. +func (c *Client) runEventsQuery(ctx context.Context, query string, startTime, endTime time.Time) (*EventsResult, error) { + started := time.Now() + + c.logger.Debug("CloudWatch Logs Insights events query", + slog.String("logGroup", c.eventsLogGroup()), + slog.String("query", query), + ) + + rows, err := c.runQuery(ctx, c.eventsLogGroup(), query, startTime, endTime) + if err != nil { + if isResourceNotFound(err) { + c.logger.Debug("Events log group not found; returning empty result", + slog.String("logGroup", c.eventsLogGroup()), + ) + return &EventsResult{Events: []EventEntry{}, TotalCount: 0, Took: int(time.Since(started).Milliseconds())}, nil + } + return nil, err + } + + entries := make([]EventEntry, 0, len(rows)) + for _, row := range rows { + entry := EventEntry{ + Message: row["message"], + Type: row["type"], + Reason: row["reason"], + ObjectKind: row["objectKind"], + ObjectName: row["objectName"], + ObjectNamespace: row["objectNamespace"], + ComponentUID: row["componentUid"], + ComponentName: row["componentName"], + EnvironmentUID: row["environmentUid"], + EnvironmentName: row["environmentName"], + ProjectUID: row["projectUid"], + ProjectName: row["projectName"], + NamespaceName: row["namespaceName"], + } + if ts, err := parseInsightsTimestamp(row["@timestamp"]); err == nil { + entry.Timestamp = ts + } + entries = append(entries, entry) + } + + return &EventsResult{ + Events: entries, + TotalCount: len(entries), + Took: int(time.Since(started).Milliseconds()), + }, nil +} + +// isResourceNotFound reports whether err is (or wraps) a CloudWatch Logs +// ResourceNotFoundException — e.g. the events log group has not been created because +// the events collector is not deployed. +func isResourceNotFound(err error) bool { + var notFound *cwltypes.ResourceNotFoundException + return errors.As(err, ¬Found) +} diff --git a/observability-logs-aws-cloudwatch/internal/cloudwatch/events_test.go b/observability-logs-aws-cloudwatch/internal/cloudwatch/events_test.go new file mode 100644 index 00000000..6b7d1d9a --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/cloudwatch/events_test.go @@ -0,0 +1,107 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package cloudwatch + +import ( + "context" + "errors" + "io" + "log/slog" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + cwltypes "github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types" +) + +func newEventsTestClient(api *queryStubLogsAPI) *Client { + return NewClientWithAWS(api, &stubAlarmsAPI{}, &stsStub{}, Config{ + LogGroupName: "/aws/containerinsights/application", + EventsLogGroupName: "/aws/containerinsights/events", + QueryTimeout: 2 * time.Second, + PollEvery: 5 * time.Millisecond, + }, slog.New(slog.NewTextHandler(io.Discard, nil))) +} + +func TestGetComponentEventsMapsAliasedColumns(t *testing.T) { + now := time.Now().UTC().Truncate(time.Second) + api := &queryStubLogsAPI{ + resultsQueue: [][]cwltypes.ResultField{ + { + {Field: aws.String("@timestamp"), Value: aws.String(now.Format("2006-01-02 15:04:05.000"))}, + {Field: aws.String("message"), Value: aws.String("Back-off restarting failed container")}, + {Field: aws.String("type"), Value: aws.String("Warning")}, + {Field: aws.String("reason"), Value: aws.String("BackOff")}, + {Field: aws.String("objectKind"), Value: aws.String("Pod")}, + {Field: aws.String("objectName"), Value: aws.String("reading-list-abc")}, + {Field: aws.String("objectNamespace"), Value: aws.String("dp-default")}, + {Field: aws.String("componentUid"), Value: aws.String("33333333-3333-3333-3333-333333333333")}, + {Field: aws.String("componentName"), Value: aws.String("reading-list")}, + {Field: aws.String("namespaceName"), Value: aws.String("default")}, + {Field: aws.String("@ptr"), Value: aws.String("ignore-me")}, + }, + }, + } + c := newEventsTestClient(api) + res, err := c.GetComponentEvents(context.Background(), ComponentEventsParams{ + Namespace: "default", + StartTime: now.Add(-time.Hour), + EndTime: now, + }) + if err != nil { + t.Fatalf("GetComponentEvents() error = %v", err) + } + if len(res.Events) != 1 || res.TotalCount != 1 { + t.Fatalf("expected 1 event, got %#v", res) + } + got := res.Events[0] + if got.Message != "Back-off restarting failed container" || got.Type != "Warning" || got.Reason != "BackOff" { + t.Fatalf("unexpected event fields: %#v", got) + } + if got.ObjectKind != "Pod" || got.ObjectName != "reading-list-abc" || got.ObjectNamespace != "dp-default" { + t.Fatalf("unexpected object metadata: %#v", got) + } + if got.ComponentName != "reading-list" || got.ComponentUID != "33333333-3333-3333-3333-333333333333" || got.NamespaceName != "default" { + t.Fatalf("unexpected scope metadata: %#v", got) + } + if !got.Timestamp.Equal(now) { + t.Fatalf("unexpected timestamp: %s", got.Timestamp) + } +} + +// TestGetComponentEventsMissingLogGroupDegradesGracefully asserts that a missing +// events log group (collector not deployed) yields an empty result, not an error. +func TestGetComponentEventsMissingLogGroupDegradesGracefully(t *testing.T) { + now := time.Now().UTC() + api := &queryStubLogsAPI{ + startErr: &cwltypes.ResourceNotFoundException{Message: aws.String("log group does not exist")}, + } + c := newEventsTestClient(api) + res, err := c.GetComponentEvents(context.Background(), ComponentEventsParams{ + Namespace: "default", + StartTime: now.Add(-time.Hour), + EndTime: now, + }) + if err != nil { + t.Fatalf("expected graceful empty result, got error = %v", err) + } + if res == nil || len(res.Events) != 0 || res.TotalCount != 0 { + t.Fatalf("expected empty result, got %#v", res) + } +} + +func TestGetWorkflowEventsPropagatesNonNotFoundErrors(t *testing.T) { + now := time.Now().UTC() + api := &queryStubLogsAPI{startErr: errors.New("throttled")} + c := newEventsTestClient(api) + _, err := c.GetWorkflowEvents(context.Background(), WorkflowEventsParams{ + Namespace: "default", + WorkflowRunName: "build-1", + StartTime: now.Add(-time.Hour), + EndTime: now, + }) + if err == nil { + t.Fatal("expected non-NotFound error to propagate") + } +} diff --git a/observability-logs-aws-cloudwatch/internal/config.go b/observability-logs-aws-cloudwatch/internal/config.go index db96d003..873d8901 100644 --- a/observability-logs-aws-cloudwatch/internal/config.go +++ b/observability-logs-aws-cloudwatch/internal/config.go @@ -13,13 +13,14 @@ import ( ) type Config struct { - ServerPort string - AWSRegion string - LogGroupPrefix string - LogGroupName string - QueryTimeout time.Duration - QueryPollEvery time.Duration - LogLevel slog.Level + ServerPort string + AWSRegion string + LogGroupPrefix string + LogGroupName string + EventsLogGroupName string + QueryTimeout time.Duration + QueryPollEvery time.Duration + LogLevel slog.Level // Alerting configuration AlertMetricNamespace string @@ -39,6 +40,7 @@ func LoadConfig() (*Config, error) { awsRegion := getEnv("AWS_REGION", "") logGroupPrefix := getEnv("LOG_GROUP_PREFIX", "/aws/containerinsights") logGroupName := getEnv("LOG_GROUP_NAME", "") + eventsLogGroupName := getEnv("EVENTS_LOG_GROUP_NAME", "") queryTimeoutStr := getEnv("QUERY_TIMEOUT_SECONDS", "30") queryPollStr := getEnv("QUERY_POLL_MILLISECONDS", "500") @@ -64,6 +66,9 @@ func LoadConfig() (*Config, error) { if logGroupName == "" { logGroupName = logGroupPrefix + "/application" } + if eventsLogGroupName == "" { + eventsLogGroupName = logGroupPrefix + "/events" + } queryTimeoutSec, err := strconv.Atoi(queryTimeoutStr) if err != nil || queryTimeoutSec <= 0 { @@ -110,6 +115,7 @@ func LoadConfig() (*Config, error) { AWSRegion: awsRegion, LogGroupPrefix: logGroupPrefix, LogGroupName: logGroupName, + EventsLogGroupName: eventsLogGroupName, QueryTimeout: time.Duration(queryTimeoutSec) * time.Second, QueryPollEvery: time.Duration(queryPollMs) * time.Millisecond, LogLevel: logLevel, diff --git a/observability-logs-aws-cloudwatch/internal/config_test.go b/observability-logs-aws-cloudwatch/internal/config_test.go index 557fbd70..39a360e5 100644 --- a/observability-logs-aws-cloudwatch/internal/config_test.go +++ b/observability-logs-aws-cloudwatch/internal/config_test.go @@ -81,6 +81,7 @@ func resetCoreEnv(t *testing.T) { t.Setenv("LOG_LEVEL", "") t.Setenv("LOG_GROUP_NAME", "") t.Setenv("LOG_GROUP_PREFIX", "") + t.Setenv("EVENTS_LOG_GROUP_NAME", "") t.Setenv("ALARM_ACTION_ARNS", "") t.Setenv("OK_ACTION_ARNS", "") t.Setenv("INSUFFICIENT_DATA_ACTION_ARNS", "") @@ -124,6 +125,34 @@ func TestLoadConfigDefaultsAreApplied(t *testing.T) { if cfg.LogGroupName != "/aws/containerinsights/application" { t.Fatalf("LogGroupName = %q", cfg.LogGroupName) } + if cfg.EventsLogGroupName != "/aws/containerinsights/events" { + t.Fatalf("EventsLogGroupName = %q", cfg.EventsLogGroupName) + } +} + +func TestLoadConfigEventsLogGroupOverrides(t *testing.T) { + resetCoreEnv(t) + t.Setenv("AWS_REGION", "eu-north-1") + + // A custom prefix flows through to the derived events log group name. + t.Setenv("LOG_GROUP_PREFIX", "/openchoreo/logs/") + cfg, err := LoadConfig() + if err != nil { + t.Fatalf("LoadConfig() error = %v", err) + } + if cfg.EventsLogGroupName != "/openchoreo/logs/events" { + t.Fatalf("derived EventsLogGroupName = %q", cfg.EventsLogGroupName) + } + + // An explicit override wins over the derived default. + t.Setenv("EVENTS_LOG_GROUP_NAME", "/custom/events-group") + cfg, err = LoadConfig() + if err != nil { + t.Fatalf("LoadConfig() error = %v", err) + } + if cfg.EventsLogGroupName != "/custom/events-group" { + t.Fatalf("override EventsLogGroupName = %q", cfg.EventsLogGroupName) + } } func TestLoadConfigRequiresAWSRegion(t *testing.T) { diff --git a/observability-logs-aws-cloudwatch/internal/event_handlers_test.go b/observability-logs-aws-cloudwatch/internal/event_handlers_test.go new file mode 100644 index 00000000..7f46a339 --- /dev/null +++ b/observability-logs-aws-cloudwatch/internal/event_handlers_test.go @@ -0,0 +1,225 @@ +// Copyright 2026 The OpenChoreo Authors +// SPDX-License-Identifier: Apache-2.0 + +package app + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/openchoreo/community-modules/observability-logs-aws-cloudwatch/internal/api/gen" + "github.com/openchoreo/community-modules/observability-logs-aws-cloudwatch/internal/cloudwatch" +) + +func makeComponentEventsScopeBody(t *testing.T, scope gen.ComponentSearchScope) gen.EventsQueryRequest_SearchScope { + t.Helper() + var s gen.EventsQueryRequest_SearchScope + if err := s.FromComponentSearchScope(scope); err != nil { + t.Fatalf("FromComponentSearchScope: %v", err) + } + return s +} + +func makeWorkflowEventsScopeBody(t *testing.T, scope gen.WorkflowSearchScope) gen.EventsQueryRequest_SearchScope { + t.Helper() + var s gen.EventsQueryRequest_SearchScope + if err := s.FromWorkflowSearchScope(scope); err != nil { + t.Fatalf("FromWorkflowSearchScope: %v", err) + } + return s +} + +func TestQueryEventsRejectsNilBody(t *testing.T) { + handler := newTestHandler(&queryStubClient{}, nil) + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + if _, ok := resp.(gen.QueryEvents400JSONResponse); !ok { + t.Fatalf("expected 400, got %T", resp) + } +} + +func TestQueryEventsRejectsEmptyComponentNamespace(t *testing.T) { + handler := newTestHandler(&queryStubClient{}, nil) + scope := makeComponentEventsScopeBody(t, gen.ComponentSearchScope{Namespace: ""}) + body := &gen.EventsQueryRequest{ + StartTime: time.Now().Add(-time.Hour), + EndTime: time.Now(), + SearchScope: scope, + } + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + if _, ok := resp.(gen.QueryEvents400JSONResponse); !ok { + t.Fatalf("expected 400, got %T", resp) + } +} + +func TestQueryEventsRejectsWorkflowWithoutRunName(t *testing.T) { + handler := newTestHandler(&queryStubClient{}, nil) + emptyRun := "" + scope := makeWorkflowEventsScopeBody(t, gen.WorkflowSearchScope{Namespace: "default", WorkflowRunName: &emptyRun}) + body := &gen.EventsQueryRequest{ + StartTime: time.Now().Add(-time.Hour), + EndTime: time.Now(), + SearchScope: scope, + } + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + if _, ok := resp.(gen.QueryEvents400JSONResponse); !ok { + t.Fatalf("expected 400, got %T", resp) + } +} + +func TestQueryEventsComponentReturnsResults(t *testing.T) { + now := time.Now().UTC() + client := &queryStubClient{ + componentEventsResult: &cloudwatch.EventsResult{ + Events: []cloudwatch.EventEntry{{ + Timestamp: now, + Message: "Back-off restarting failed container", + Type: "Warning", + Reason: "BackOff", + ObjectKind: "Pod", + ObjectName: "reading-list-abc", + ObjectNamespace: "dp-default", + ComponentUID: "33333333-3333-3333-3333-333333333333", + ComponentName: "reading-list", + NamespaceName: "default", + }}, + TotalCount: 1, + Took: 7, + }, + } + handler := newTestHandler(client, nil) + componentUID := "33333333-3333-3333-3333-333333333333" + scope := makeComponentEventsScopeBody(t, gen.ComponentSearchScope{ + Namespace: "default", + ComponentUid: &componentUID, + }) + body := &gen.EventsQueryRequest{ + StartTime: now.Add(-time.Hour), + EndTime: now, + SearchScope: scope, + } + + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + ok, isOK := resp.(gen.QueryEvents200JSONResponse) + if !isOK { + t.Fatalf("expected 200, got %T", resp) + } + if ok.Total == nil || *ok.Total != 1 { + t.Fatalf("unexpected total: %#v", ok.Total) + } + if ok.Events == nil || len(*ok.Events) != 1 { + t.Fatalf("expected 1 event, got %#v", ok.Events) + } + got := (*ok.Events)[0] + if got.Type == nil || *got.Type != "Warning" || got.Reason == nil || *got.Reason != "BackOff" { + t.Fatalf("unexpected event: %#v", got) + } + if got.Metadata == nil || got.Metadata.ObjectKind == nil || *got.Metadata.ObjectKind != "Pod" { + t.Fatalf("unexpected metadata: %#v", got.Metadata) + } + if got.Metadata.ComponentUid == nil { + t.Fatalf("expected component UID to be parsed") + } + // The component scope must reach the client unchanged. + if client.lastComponentEvents == nil || client.lastComponentEvents.ComponentID != componentUID { + t.Fatalf("unexpected component params: %#v", client.lastComponentEvents) + } +} + +func TestQueryEventsWorkflowReturnsResults(t *testing.T) { + now := time.Now().UTC() + client := &queryStubClient{ + workflowEventsResult: &cloudwatch.EventsResult{ + Events: []cloudwatch.EventEntry{{ + Timestamp: now, + Message: "Created pod", + Type: "Normal", + Reason: "SuccessfulCreate", + ObjectKind: "Job", + ObjectName: "build-123-clone", + }}, + TotalCount: 1, + }, + } + handler := newTestHandler(client, nil) + runName := "build-123" + taskName := "clone" + scope := makeWorkflowEventsScopeBody(t, gen.WorkflowSearchScope{ + Namespace: "default", + WorkflowRunName: &runName, + TaskName: &taskName, + }) + body := &gen.EventsQueryRequest{ + StartTime: now.Add(-time.Hour), + EndTime: now, + SearchScope: scope, + } + + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + if _, ok := resp.(gen.QueryEvents200JSONResponse); !ok { + t.Fatalf("expected 200, got %T", resp) + } + if client.lastWorkflowEvents == nil || client.lastWorkflowEvents.WorkflowRunName != runName || client.lastWorkflowEvents.TaskName != taskName { + t.Fatalf("unexpected workflow params: %#v", client.lastWorkflowEvents) + } +} + +func TestQueryEventsComponentReturnsServerErrorOnClientFailure(t *testing.T) { + handler := newTestHandler(&queryStubClient{componentEventsErr: errors.New("aws boom")}, nil) + scope := makeComponentEventsScopeBody(t, gen.ComponentSearchScope{Namespace: "default"}) + body := &gen.EventsQueryRequest{ + StartTime: time.Now().Add(-time.Hour), + EndTime: time.Now(), + SearchScope: scope, + } + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + if _, ok := resp.(gen.QueryEvents500JSONResponse); !ok { + t.Fatalf("expected 500, got %T", resp) + } +} + +// A missing events log group is surfaced by the client as an empty result (not an +// error), so the handler must return 200 with an empty events array. +func TestQueryEventsComponentEmptyResultReturns200(t *testing.T) { + client := &queryStubClient{componentEventsResult: &cloudwatch.EventsResult{Events: []cloudwatch.EventEntry{}, TotalCount: 0}} + handler := newTestHandler(client, nil) + scope := makeComponentEventsScopeBody(t, gen.ComponentSearchScope{Namespace: "default"}) + body := &gen.EventsQueryRequest{ + StartTime: time.Now().Add(-time.Hour), + EndTime: time.Now(), + SearchScope: scope, + } + resp, err := handler.QueryEvents(context.Background(), gen.QueryEventsRequestObject{Body: body}) + if err != nil { + t.Fatalf("QueryEvents() error = %v", err) + } + ok, isOK := resp.(gen.QueryEvents200JSONResponse) + if !isOK { + t.Fatalf("expected 200, got %T", resp) + } + if ok.Total == nil || *ok.Total != 0 { + t.Fatalf("expected total 0, got %#v", ok.Total) + } + if ok.Events == nil || len(*ok.Events) != 0 { + t.Fatalf("expected empty events array, got %#v", ok.Events) + } +} diff --git a/observability-logs-aws-cloudwatch/internal/handlers.go b/observability-logs-aws-cloudwatch/internal/handlers.go index 5ec1eed5..d2044769 100644 --- a/observability-logs-aws-cloudwatch/internal/handlers.go +++ b/observability-logs-aws-cloudwatch/internal/handlers.go @@ -25,6 +25,8 @@ type logsClient interface { Ping(context.Context) error GetComponentLogs(context.Context, cloudwatch.ComponentLogsParams) (*cloudwatch.ComponentLogsResult, error) GetWorkflowLogs(context.Context, cloudwatch.WorkflowLogsParams) (*cloudwatch.WorkflowLogsResult, error) + GetComponentEvents(context.Context, cloudwatch.ComponentEventsParams) (*cloudwatch.EventsResult, error) + GetWorkflowEvents(context.Context, cloudwatch.WorkflowEventsParams) (*cloudwatch.EventsResult, error) CreateAlert(context.Context, cloudwatch.LogAlertParams) (string, error) GetAlert(context.Context, string, string) (*cloudwatch.AlertDetail, error) UpdateAlert(context.Context, string, string, cloudwatch.LogAlertParams) (string, error) @@ -161,6 +163,67 @@ func (h *LogsHandler) QueryLogs(ctx context.Context, request gen.QueryLogsReques return gen.QueryLogs200JSONResponse(toComponentLogsQueryResponse(result)), nil } +// QueryEvents implements POST /api/v1/events/query. +func (h *LogsHandler) QueryEvents(ctx context.Context, request gen.QueryEventsRequestObject) (gen.QueryEventsResponseObject, error) { + if request.Body == nil { + return gen.QueryEvents400JSONResponse{ + Title: ptr(gen.BadRequest), + Message: ptr("request body is required"), + }, nil + } + + // WorkflowSearchScope is identified by the presence of workflowRunName. + workflowScope, err := request.Body.SearchScope.AsWorkflowSearchScope() + if err == nil && workflowScope.WorkflowRunName != nil { + if strings.TrimSpace(workflowScope.Namespace) == "" || strings.TrimSpace(*workflowScope.WorkflowRunName) == "" { + return gen.QueryEvents400JSONResponse{ + Title: ptr(gen.BadRequest), + Message: ptr("searchScope with a valid namespace and workflowRunName is required"), + }, nil + } + + params := toWorkflowEventsParams(request.Body, &workflowScope) + result, err := h.client.GetWorkflowEvents(ctx, params) + if err != nil { + h.logger.Error("Failed to query workflow events", + slog.String("function", "QueryEvents"), + slog.String("namespace", workflowScope.Namespace), + slog.Any("error", err), + ) + return gen.QueryEvents500JSONResponse{ + Title: ptr(gen.InternalServerError), + Message: ptr("internal server error"), + }, nil + } + + return gen.QueryEvents200JSONResponse(toEventsQueryResponse(result)), nil + } + + scope, err := request.Body.SearchScope.AsComponentSearchScope() + if err != nil || strings.TrimSpace(scope.Namespace) == "" { + return gen.QueryEvents400JSONResponse{ + Title: ptr(gen.BadRequest), + Message: ptr("searchScope with a valid namespace is required"), + }, nil + } + + params := toComponentEventsParams(request.Body, &scope) + result, err := h.client.GetComponentEvents(ctx, params) + if err != nil { + h.logger.Error("Failed to query component events", + slog.String("function", "QueryEvents"), + slog.String("namespace", scope.Namespace), + slog.Any("error", err), + ) + return gen.QueryEvents500JSONResponse{ + Title: ptr(gen.InternalServerError), + Message: ptr("internal server error"), + }, nil + } + + return gen.QueryEvents200JSONResponse(toEventsQueryResponse(result)), nil +} + // CreateAlertRule implements POST /api/v1alpha1/alerts/rules. func (h *LogsHandler) CreateAlertRule(ctx context.Context, request gen.CreateAlertRuleRequestObject) (gen.CreateAlertRuleResponseObject, error) { if request.Body == nil { @@ -755,6 +818,111 @@ func toComponentLogEntry(l *cloudwatch.ComponentLogsEntry) gen.ComponentLogEntry return entry } +func toComponentEventsParams(req *gen.EventsQueryRequest, scope *gen.ComponentSearchScope) cloudwatch.ComponentEventsParams { + params := cloudwatch.ComponentEventsParams{ + Namespace: scope.Namespace, + StartTime: req.StartTime, + EndTime: req.EndTime, + } + if scope.ProjectUid != nil { + params.ProjectID = *scope.ProjectUid + } + if scope.EnvironmentUid != nil { + params.EnvironmentID = *scope.EnvironmentUid + } + if scope.ComponentUid != nil { + params.ComponentID = *scope.ComponentUid + } + if req.Limit != nil { + params.Limit = *req.Limit + } + if req.SortOrder != nil { + params.SortOrder = string(*req.SortOrder) + } + return params +} + +func toWorkflowEventsParams(req *gen.EventsQueryRequest, scope *gen.WorkflowSearchScope) cloudwatch.WorkflowEventsParams { + params := cloudwatch.WorkflowEventsParams{ + Namespace: scope.Namespace, + StartTime: req.StartTime, + EndTime: req.EndTime, + } + if scope.WorkflowRunName != nil { + params.WorkflowRunName = *scope.WorkflowRunName + } + if scope.TaskName != nil { + params.TaskName = *scope.TaskName + } + if req.Limit != nil { + params.Limit = *req.Limit + } + if req.SortOrder != nil { + params.SortOrder = string(*req.SortOrder) + } + return params +} + +func toEventsQueryResponse(result *cloudwatch.EventsResult) gen.EventsQueryResponse { + entries := make([]gen.EventEntry, 0, len(result.Events)) + for i := range result.Events { + entries = append(entries, toEventEntry(&result.Events[i])) + } + return gen.EventsQueryResponse{ + Events: &entries, + Total: &result.TotalCount, + TookMs: &result.Took, + } +} + +func toEventEntry(e *cloudwatch.EventEntry) gen.EventEntry { + ts := e.Timestamp + entry := gen.EventEntry{ + Timestamp: &ts, + Message: strPtr(e.Message), + Type: strPtr(e.Type), + Reason: strPtr(e.Reason), + Metadata: &struct { + ComponentName *string `json:"componentName,omitempty"` + ComponentUid *openapi_types.UUID `json:"componentUid,omitempty"` + EnvironmentName *string `json:"environmentName,omitempty"` + EnvironmentUid *openapi_types.UUID `json:"environmentUid,omitempty"` + NamespaceName *string `json:"namespaceName,omitempty"` + ObjectKind *string `json:"objectKind,omitempty"` + ObjectName *string `json:"objectName,omitempty"` + ObjectNamespace *string `json:"objectNamespace,omitempty"` + ProjectName *string `json:"projectName,omitempty"` + ProjectUid *openapi_types.UUID `json:"projectUid,omitempty"` + }{ + ComponentName: strPtr(e.ComponentName), + EnvironmentName: strPtr(e.EnvironmentName), + NamespaceName: strPtr(e.NamespaceName), + ObjectKind: strPtr(e.ObjectKind), + ObjectName: strPtr(e.ObjectName), + ObjectNamespace: strPtr(e.ObjectNamespace), + ProjectName: strPtr(e.ProjectName), + }, + } + + if e.ComponentUID != "" { + if uid, ok := parseUUID(e.ComponentUID); ok { + entry.Metadata.ComponentUid = &uid + } + } + if e.ProjectUID != "" { + if uid, ok := parseUUID(e.ProjectUID); ok { + entry.Metadata.ProjectUid = &uid + } + } + if e.EnvironmentUID != "" { + if uid, ok := parseUUID(e.EnvironmentUID); ok { + entry.Metadata.EnvironmentUid = &uid + } + } + + return entry +} + func parseUUID(s string) (openapi_types.UUID, bool) { parsed, err := uuid.Parse(s) if err != nil { diff --git a/observability-logs-aws-cloudwatch/internal/handlers_test.go b/observability-logs-aws-cloudwatch/internal/handlers_test.go index 07dec2b7..19d2fbc0 100644 --- a/observability-logs-aws-cloudwatch/internal/handlers_test.go +++ b/observability-logs-aws-cloudwatch/internal/handlers_test.go @@ -32,6 +32,12 @@ func (s *stubLogsClient) GetComponentLogs(context.Context, cloudwatch.ComponentL func (s *stubLogsClient) GetWorkflowLogs(context.Context, cloudwatch.WorkflowLogsParams) (*cloudwatch.WorkflowLogsResult, error) { return nil, errors.New("unexpected GetWorkflowLogs call") } +func (s *stubLogsClient) GetComponentEvents(context.Context, cloudwatch.ComponentEventsParams) (*cloudwatch.EventsResult, error) { + return nil, errors.New("unexpected GetComponentEvents call") +} +func (s *stubLogsClient) GetWorkflowEvents(context.Context, cloudwatch.WorkflowEventsParams) (*cloudwatch.EventsResult, error) { + return nil, errors.New("unexpected GetWorkflowEvents call") +} func (s *stubLogsClient) CreateAlert(ctx context.Context, params cloudwatch.LogAlertParams) (string, error) { if s.createAlertFn == nil { return "", errors.New("unexpected CreateAlert call") @@ -354,6 +360,13 @@ type queryStubClient struct { componentErr error workflowResult *cloudwatch.WorkflowLogsResult workflowErr error + + componentEventsResult *cloudwatch.EventsResult + componentEventsErr error + workflowEventsResult *cloudwatch.EventsResult + workflowEventsErr error + lastComponentEvents *cloudwatch.ComponentEventsParams + lastWorkflowEvents *cloudwatch.WorkflowEventsParams } func (c *queryStubClient) GetComponentLogs(_ context.Context, _ cloudwatch.ComponentLogsParams) (*cloudwatch.ComponentLogsResult, error) { @@ -364,6 +377,16 @@ func (c *queryStubClient) GetWorkflowLogs(_ context.Context, _ cloudwatch.Workfl return c.workflowResult, c.workflowErr } +func (c *queryStubClient) GetComponentEvents(_ context.Context, p cloudwatch.ComponentEventsParams) (*cloudwatch.EventsResult, error) { + c.lastComponentEvents = &p + return c.componentEventsResult, c.componentEventsErr +} + +func (c *queryStubClient) GetWorkflowEvents(_ context.Context, p cloudwatch.WorkflowEventsParams) (*cloudwatch.EventsResult, error) { + c.lastWorkflowEvents = &p + return c.workflowEventsResult, c.workflowEventsErr +} + func TestHealthReturnsHealthy(t *testing.T) { handler := newTestHandler(&stubLogsClient{}, nil) resp, err := handler.Health(context.Background(), gen.HealthRequestObject{}) diff --git a/observability-logs-aws-cloudwatch/main.go b/observability-logs-aws-cloudwatch/main.go index 6d64bf24..a5e533cb 100644 --- a/observability-logs-aws-cloudwatch/main.go +++ b/observability-logs-aws-cloudwatch/main.go @@ -30,6 +30,7 @@ func main() { slog.String("logLevel", cfg.LogLevel.String()), slog.String("awsRegion", cfg.AWSRegion), slog.String("logGroupName", cfg.LogGroupName), + slog.String("eventsLogGroupName", cfg.EventsLogGroupName), slog.String("queryTimeout", cfg.QueryTimeout.String()), slog.String("serverPort", cfg.ServerPort), ) @@ -40,6 +41,7 @@ func main() { cwClient, err := cloudwatch.NewClient(bootstrapCtx, cloudwatch.Config{ Region: cfg.AWSRegion, LogGroupName: cfg.LogGroupName, + EventsLogGroupName: cfg.EventsLogGroupName, QueryTimeout: cfg.QueryTimeout, PollEvery: cfg.QueryPollEvery, AlertMetricNamespace: cfg.AlertMetricNamespace, From ce2d03b668559b83c6f5ee196834316127b94e9c Mon Sep 17 00:00:00 2001 From: Chalindu Kodikara Date: Fri, 10 Jul 2026 17:23:49 +0530 Subject: [PATCH 2/2] docs: update event publisher readme Signed-off-by: Chalindu Kodikara --- observability-events-otel-collector/README.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/observability-events-otel-collector/README.md b/observability-events-otel-collector/README.md index ad5c7358..8df9b741 100644 --- a/observability-events-otel-collector/README.md +++ b/observability-events-otel-collector/README.md @@ -149,13 +149,17 @@ EOF ### AWS CloudWatch Logs -Credentials come from the standard AWS chain (IRSA on EKS): +Compatible with `observability-logs-aws-cloudwatch` when the adapter chart is +installed with `events.enabled=true`. The adapter setup job creates the events +log group with retention; the collector only needs permission to create streams +and publish log events. Credentials come from the standard AWS chain (IRSA on +EKS): ```yaml exporters: awscloudwatchlogs: region: "us-east-1" - log_group_name: "/openchoreo/k8s-events" + log_group_name: "/aws/containerinsights/events" log_stream_name: "events" pipelineExporters: - awscloudwatchlogs