From e7d4320d2ac1818288558c4e01ca763e1a264843 Mon Sep 17 00:00:00 2001 From: Michael O'Neill Date: Mon, 9 Nov 2026 17:20:21 +0000 Subject: [PATCH] test(otel): cover the Helm chart's default-on OTLP endpoint on EKS and AKS The amazon-cloudwatch-observability chart now serves an OTLP endpoint on the node agent by default (4317 gRPC, 4318 HTTP), with Container Insights enrichment, the OpenTelemetry Kubernetes service identity rules and otelFilters applied to OTLP telemetry. Ordinary pods send through the cloudwatch-agent Service, and the agent finds the sender pod from the connection's source IP. - terraform/eks/daemon/otel-otlp + test/otel/otlp: chart defaults plus OTel Container Insights. Senders through the Service cover the identity cases (app labels, annotations, a service.name the sender sets, unknown_service, DaemonSet, CronJob). The tests check the resource attributes against the sender pod and its node, scope tags, span metrics, the exact OTLP metric name set, logs in /aws/cwagent//otlp, traces in Transaction Search, and that Container Insights telemetry is unchanged. - terraform/eks/daemon/otel-otlp-disabled + test/otel/otlp_disabled: otlp.enabled=false. No OTLP Service or host ports, a send through the Service fails, and no metric with the OTLP scope arrives. - test/otel/filters: an OTLP sender container in every workload. The namespace, pod and node label and metric name filters apply to OTLP metrics, logs and spans; a non-empty metricNames.include also drops an unlisted OTLP metric and the span metrics. - test/azure/aks containerinsights mode: a sender through the Service and checks of its metrics, span metrics, metric names, logs and traces. The install values are unchanged; default:otel is now an alias of the default config. - terraform/otlp_service_sender.sh: the shared OTLP/HTTP sender. It logs the HTTP status of every post, which the tests read. - test/otel/otlpendpoint: helpers shared by the suites, with unit tests. - The new EKS terraform takes helm_chart_repo, so a chart branch on a fork can be tested, and stops with a clear error when the chart has no otlp values block. --- generator/test_case_generator.go | 14 + terraform/azure/aks/main.tf | 28 +- terraform/eks/daemon/otel-filters/main.tf | 40 +- .../eks/daemon/otel-otlp-disabled/main.tf | 300 ++++++++++ .../daemon/otel-otlp-disabled/providers.tf | 28 + .../daemon/otel-otlp-disabled/variables.tf | 49 ++ terraform/eks/daemon/otel-otlp/main.tf | 522 +++++++++++++++++ terraform/eks/daemon/otel-otlp/providers.tf | 28 + terraform/eks/daemon/otel-otlp/variables.tf | 49 ++ terraform/otlp_service_sender.sh | 68 +++ .../aks/containerinsights/aks_otlp_test.go | 225 ++++++++ .../resources/otlp_sender.yaml | 41 ++ test/otel/filters/filters_test.go | 39 +- test/otel/filters/otlp_test.go | 172 ++++++ test/otel/otlp/otlp_test.go | 541 ++++++++++++++++++ test/otel/otlp/setup_test.go | 111 ++++ test/otel/otlp_disabled/otlp_disabled_test.go | 87 +++ test/otel/otlp_disabled/setup_test.go | 92 +++ test/otel/otlpendpoint/otlpendpoint.go | 158 +++++ test/otel/otlpendpoint/otlpendpoint_test.go | 40 ++ 20 files changed, 2613 insertions(+), 19 deletions(-) create mode 100644 terraform/eks/daemon/otel-otlp-disabled/main.tf create mode 100644 terraform/eks/daemon/otel-otlp-disabled/providers.tf create mode 100644 terraform/eks/daemon/otel-otlp-disabled/variables.tf create mode 100644 terraform/eks/daemon/otel-otlp/main.tf create mode 100644 terraform/eks/daemon/otel-otlp/providers.tf create mode 100644 terraform/eks/daemon/otel-otlp/variables.tf create mode 100644 terraform/otlp_service_sender.sh create mode 100644 test/azure/aks/containerinsights/aks_otlp_test.go create mode 100644 test/azure/aks/containerinsights/resources/otlp_sender.yaml create mode 100644 test/otel/filters/otlp_test.go create mode 100644 test/otel/otlp/otlp_test.go create mode 100644 test/otel/otlp/setup_test.go create mode 100644 test/otel/otlp_disabled/otlp_disabled_test.go create mode 100644 test/otel/otlp_disabled/setup_test.go create mode 100644 test/otel/otlpendpoint/otlpendpoint.go create mode 100644 test/otel/otlpendpoint/otlpendpoint_test.go diff --git a/generator/test_case_generator.go b/generator/test_case_generator.go index 4d446d505..57696981c 100644 --- a/generator/test_case_generator.go +++ b/generator/test_case_generator.go @@ -496,6 +496,20 @@ var testTypeToTestConfig = map[string][]testConfig{ ami: "AL2023_x86_64_STANDARD", k8sVersion: "1.35", }, + { + testDir: "./test/otel/otlp", + terraformDir: "terraform/eks/daemon/otel-otlp", + targets: map[string]map[string]struct{}{"arc": {"amd64": {}}}, + ami: "AL2023_x86_64_STANDARD", + k8sVersion: "1.35", + }, + { + testDir: "./test/otel/otlp_disabled", + terraformDir: "terraform/eks/daemon/otel-otlp-disabled", + targets: map[string]map[string]struct{}{"arc": {"amd64": {}}}, + ami: "AL2023_x86_64_STANDARD", + k8sVersion: "1.35", + }, { testDir: "./test/otel/ebs_csi", terraformDir: "terraform/eks/daemon/otel-ebs-csi", diff --git a/terraform/azure/aks/main.tf b/terraform/azure/aks/main.tf index 4ea5932ef..9ac816f4a 100644 --- a/terraform/azure/aks/main.tf +++ b/terraform/azure/aks/main.tf @@ -407,7 +407,7 @@ resource "null_resource" "agent_diagnostics" { EOT } - depends_on = [kubernetes_daemon_set_v1.cwagent, kubernetes_job_v1.otlp_load, null_resource.ci_helm_chart] + depends_on = [kubernetes_daemon_set_v1.cwagent, kubernetes_job_v1.otlp_load, null_resource.ci_helm_chart, null_resource.ci_otlp_sender] } ##################################################################### @@ -433,7 +433,7 @@ resource "null_resource" "integration_test" { } } - depends_on = [kubernetes_daemon_set_v1.cwagent, kubernetes_job_v1.otlp_load, null_resource.ci_helm_chart, null_resource.ci_extra_manifests, null_resource.agent_diagnostics] + depends_on = [kubernetes_daemon_set_v1.cwagent, kubernetes_job_v1.otlp_load, null_resource.ci_helm_chart, null_resource.ci_extra_manifests, null_resource.ci_otlp_sender, null_resource.agent_diagnostics] } ##################################################################### @@ -513,6 +513,30 @@ resource "null_resource" "ci_helm_chart" { ] } +##################################################################### +# OTLP sender (test_mode == "containerinsights"): an ordinary pod that +# sends OTLP through the cloudwatch-agent Service to the chart's OTLP +# endpoint, which the chart serves by default. default:otel in the +# install values above is an alias of the default config, so the +# endpoint comes from the chart, not from the agent's default:otel. +##################################################################### +resource "null_resource" "ci_otlp_sender" { + count = local.is_ci ? 1 : 0 + provisioner "local-exec" { + command = <<-EOT + set -euo pipefail + KC='${local_sensitive_file.kubeconfig.filename}' + kubectl --kubeconfig "$KC" create namespace otlp-e2e --dry-run=client -o yaml | kubectl --kubeconfig "$KC" apply -f - + kubectl --kubeconfig "$KC" -n otlp-e2e create configmap otlp-e2e-sender \ + --from-file=sender.sh='${path.module}/../../otlp_service_sender.sh' --dry-run=client -o yaml | kubectl --kubeconfig "$KC" apply -f - + kubectl --kubeconfig "$KC" apply -f '${path.module}/../../../${var.test_dir}/resources/otlp_sender.yaml' + kubectl --kubeconfig "$KC" -n otlp-e2e rollout status deployment/otlp-e2e-app --timeout=5m + EOT + interpreter = ["/bin/bash", "-c"] + } + depends_on = [null_resource.ci_helm_chart] +} + ##################################################################### # keda/karpenter stub emitters + sample app (applied via kubectl). ##################################################################### diff --git a/terraform/eks/daemon/otel-filters/main.tf b/terraform/eks/daemon/otel-filters/main.tf index 6f103a66e..d5945eb15 100644 --- a/terraform/eks/daemon/otel-filters/main.tf +++ b/terraform/eks/daemon/otel-filters/main.tf @@ -35,6 +35,9 @@ locals { # Namespaces of the stand-in inference servers: one kept and one dropped by # the namespaces filter. inference_namespaces = ["filters-keep-a", "filters-keep-secret-b"] + + # Run by the OTLP sender container of each test workload. + sender_script = "${path.module}/../../../otlp_service_sender.sh" } resource "aws_eks_cluster" "this" { @@ -173,6 +176,9 @@ resource "helm_release" "aws_observability" { # the namespaces filter alone decides whether they arrive. Both solutions # are on by default. The values set them so that the test does not depend # on the default. + # metricNames.include names otlp_e2e_requests, one of the two counters the + # OTLP sender container sends, so the other one, otlp_e2e_unlisted, and the + # span metrics computed from the sender's spans are dropped. # filters-e2e.example.com/keep* is the only include entry that covers the pod # label filters-e2e.example.com/keep-x, so it tests a dotted prefix include. # The pod labels app-extra and filters-e2e-keep-secret-x and the metric @@ -192,7 +198,7 @@ resource "helm_release" "aws_observability" { include: ["filters-keep-*", "filters-exact"] exclude: ["filters-keep-secret*"] metricNames: - include: ["container_*", "kube_pod_info", "k8s.pod.*", "node_load1", "vllm:num_requests_running", "revision_request_count"] + include: ["container_*", "kube_pod_info", "k8s.pod.*", "node_load1", "vllm:num_requests_running", "revision_request_count", "otlp_e2e_requests"] exclude: ["container_network_*", "container_memory_usage_bytes"] podLabels: include: ["app", "filters-e2e-keep*", "filters-e2e.example.com/keep*", "pod-template-hash"] @@ -255,18 +261,21 @@ resource "null_resource" "restart_pods" { } # --- Test workloads: one logging deployment per namespace, one pod per node --- +# Each pod also runs an OTLP sender container, which posts OTLP through the +# cloudwatch-agent Service (terraform/otlp_service_sender.sh), so the OTLP +# endpoint's telemetry from each namespace carries the same pod labels as the +# Container Insights telemetry. resource "null_resource" "workloads" { for_each = toset(local.namespaces) depends_on = [aws_eks_node_group.this, null_resource.kubectl] provisioner "local-exec" { command = <<-EOT + set -e + kubectl create namespace ${each.key} --dry-run=client -o yaml | kubectl apply -f - + kubectl -n ${each.key} create configmap otlp-e2e-sender \ + --from-file=sender.sh=${local.sender_script} --dry-run=client -o yaml | kubectl apply -f - cat <<'EOF' | kubectl apply -f - - apiVersion: v1 - kind: Namespace - metadata: - name: ${each.key} - --- apiVersion: apps/v1 kind: Deployment metadata: @@ -304,6 +313,25 @@ resource "null_resource" "workloads" { requests: cpu: 10m memory: 16Mi + - name: otlp-sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: filters + - name: OTLP_METRICS + value: otlp_e2e_requests otlp_e2e_unlisted + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender EOF kubectl -n ${each.key} rollout status deployment/filters-e2e --timeout=5m EOT diff --git a/terraform/eks/daemon/otel-otlp-disabled/main.tf b/terraform/eks/daemon/otel-otlp-disabled/main.tf new file mode 100644 index 000000000..9c4825436 --- /dev/null +++ b/terraform/eks/daemon/otel-otlp-disabled/main.tf @@ -0,0 +1,300 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT + +# The chart with otlp.enabled=false, next to OTel Container Insights: the node +# agent must serve no OTLP endpoint. An ordinary pod tries to send OTLP through +# the cloudwatch-agent Service and logs the result, which test/otel/otlp_disabled +# reads. + +module "common" { + source = "../../../common" + cwagent_image_repo = var.cwagent_image_repo + cwagent_image_tag = var.cwagent_image_tag +} + +module "basic_components" { + source = "../../../basic_components" + region = var.region +} + +locals { + aws_eks = "aws eks --region ${var.region}" + + sender_script = "${path.module}/../../../otlp_service_sender.sh" +} + +resource "aws_eks_cluster" "this" { + name = "cwagent-eks-integ-${module.common.testing_id}" + role_arn = module.basic_components.role_arn + version = var.k8s_version + vpc_config { + subnet_ids = module.basic_components.public_subnet_ids + security_group_ids = [module.basic_components.security_group] + } +} + +resource "aws_eks_node_group" "this" { + cluster_name = aws_eks_cluster.this.name + node_group_name = "cwagent-otlp-off-${module.common.testing_id}" + node_role_arn = aws_iam_role.node_role.arn + subnet_ids = module.basic_components.public_subnet_ids + + scaling_config { + desired_size = 1 + max_size = 1 + min_size = 1 + } + + ami_type = var.ami_type + capacity_type = "ON_DEMAND" + disk_size = 20 + instance_types = [var.instance_type] + + depends_on = [ + aws_iam_role_policy_attachment.node_AmazonEC2ContainerRegistryReadOnly, + aws_iam_role_policy_attachment.node_AmazonEKS_CNI_Policy, + aws_iam_role_policy_attachment.node_AmazonEKSWorkerNodePolicy, + ] +} + +# EKS Node IAM Role +resource "aws_iam_role" "node_role" { + name = "cwagent-otlp-off-eks-Worker-Role-${module.common.testing_id}" + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [{ + Effect = "Allow" + Principal = { Service = "ec2.amazonaws.com" } + Action = "sts:AssumeRole" + }] + }) +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEKSWorkerNodePolicy" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy" + role = aws_iam_role.node_role.name +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEKS_CNI_Policy" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy" + role = aws_iam_role.node_role.name +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEC2ContainerRegistryReadOnly" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly" + role = aws_iam_role.node_role.name +} + +# Pod Identity IAM Role. CloudWatchAgentServerPolicy is the policy customers +# are told to use, and covers the OTLP metrics, logs and traces exports. +resource "aws_iam_role" "pod_identity_role" { + name = "cwagent-otlp-off-pod-identity-${module.common.testing_id}" + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [{ + Effect = "Allow" + Principal = { Service = "pods.eks.amazonaws.com" } + Action = ["sts:AssumeRole", "sts:TagSession"] + }] + }) +} + +resource "aws_iam_role_policy_attachment" "pod_identity_CloudWatchAgentServerPolicy" { + policy_arn = "arn:aws:iam::aws:policy/CloudWatchAgentServerPolicy" + role = aws_iam_role.pod_identity_role.name +} + +# --- EKS Addon: Pod Identity agent --- + +resource "aws_eks_addon" "pod_identity_agent" { + depends_on = [aws_eks_node_group.this] + cluster_name = aws_eks_cluster.this.name + addon_name = "eks-pod-identity-agent" +} + +# --- Update kubeconfig --- + +resource "null_resource" "kubectl" { + depends_on = [aws_eks_cluster.this, aws_eks_node_group.this] + provisioner "local-exec" { + command = "${local.aws_eks} update-kubeconfig --name ${aws_eks_cluster.this.name}" + } +} + +# --- Helm chart install --- + +data "external" "clone_helm_chart" { + program = ["bash", "-c", <<-EOT + set -e + rm -rf ./helm-charts + git clone -b ${var.helm_chart_branch} ${var.helm_chart_repo} ./helm-charts >&2 + # A chart without the OTLP endpoint would install fine and fail every + # test; stop here with the reason instead. + if ! grep -qE '^otlp:' ./helm-charts/charts/amazon-cloudwatch-observability/values.yaml; then + echo "${var.helm_chart_repo}@${var.helm_chart_branch} has no top-level otlp block in values.yaml" >&2 + exit 1 + fi + echo '{"status":"ready"}' + EOT + ] +} + +# The chart defaults with the OTLP endpoint turned off, plus OTel Container +# Insights, which must not bring an OTLP receiver back. +resource "helm_release" "aws_observability" { + name = "amazon-cloudwatch-observability" + chart = "./helm-charts/charts/amazon-cloudwatch-observability" + namespace = "amazon-cloudwatch" + create_namespace = true + + set = [ + { name = "clusterName", value = aws_eks_cluster.this.name }, + { name = "region", value = var.region }, + { name = "otelContainerInsights.enabled", value = "true" }, + { name = "otlp.enabled", value = "false" }, + ] + + depends_on = [ + aws_eks_addon.pod_identity_agent, + null_resource.kubectl, + data.external.clone_helm_chart, + ] +} + +# --- Pod Identity association (after Helm creates the service account) --- + +resource "aws_eks_pod_identity_association" "cloudwatch_agent" { + depends_on = [helm_release.aws_observability] + cluster_name = aws_eks_cluster.this.name + namespace = "amazon-cloudwatch" + service_account = "cloudwatch-agent" + role_arn = aws_iam_role.pod_identity_role.arn +} + +# --- Patch agent image (both DaemonSet and cluster-scraper) --- + +resource "null_resource" "update_image" { + depends_on = [helm_release.aws_observability, null_resource.kubectl] + triggers = { timestamp = timestamp() } + provisioner "local-exec" { + command = <<-EOT + set -e + sleep 30 + kubectl -n amazon-cloudwatch patch AmazonCloudWatchAgent cloudwatch-agent --type='json' \ + -p='[{"op": "replace", "path": "/spec/image", "value": "${var.cwagent_image_repo}:${var.cwagent_image_tag}"}]' + kubectl -n amazon-cloudwatch patch AmazonCloudWatchAgent cloudwatch-agent-cluster-scraper --type='json' \ + -p='[{"op": "replace", "path": "/spec/image", "value": "${var.cwagent_image_repo}:${var.cwagent_image_tag}"}]' 2>/dev/null || true + sleep 10 + EOT + } +} + +# --- Restart pods to pick up Pod Identity + new image --- + +resource "null_resource" "restart_pods" { + depends_on = [aws_eks_pod_identity_association.cloudwatch_agent, null_resource.update_image] + triggers = { timestamp = timestamp() } + provisioner "local-exec" { + command = <<-EOT + set -e + kubectl -n amazon-cloudwatch rollout restart daemonset/cloudwatch-agent + kubectl -n amazon-cloudwatch rollout restart deployment/cloudwatch-agent-cluster-scraper 2>/dev/null || true + kubectl -n amazon-cloudwatch rollout status daemonset/cloudwatch-agent --timeout=120s + kubectl -n amazon-cloudwatch rollout status deployment/cloudwatch-agent-cluster-scraper --timeout=120s 2>/dev/null || true + EOT + } +} + +# --- OTLP probe --- +# An ordinary pod that sends OTLP to the default endpoint, +# http://cloudwatch-agent.amazon-cloudwatch:4318, and logs the HTTP status of +# every post (000 when the connection fails). + +resource "null_resource" "senders" { + depends_on = [null_resource.restart_pods] + provisioner "local-exec" { + command = <<-EOT + set -e + kubectl create namespace otlp-e2e --dry-run=client -o yaml | kubectl apply -f - + kubectl -n otlp-e2e create configmap otlp-e2e-sender \ + --from-file=sender.sh=${local.sender_script} --dry-run=client -o yaml | kubectl apply -f - + cat <<'EOF' | kubectl apply -f - + apiVersion: apps/v1 + kind: Deployment + metadata: + name: otlp-e2e-probe + namespace: otlp-e2e + spec: + replicas: 1 + selector: + matchLabels: + app: otlp-e2e-probe + template: + metadata: + labels: + app: otlp-e2e-probe + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: disabled + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + EOF + kubectl -n otlp-e2e rollout status deployment/otlp-e2e-probe --timeout=5m + EOT + } +} + +# --- Diagnostics: agent and sender state in the job output, so a failed run +# can be debugged after the cluster is destroyed --- + +resource "null_resource" "agent_diagnostics" { + depends_on = [null_resource.senders] + triggers = { always_run = timestamp() } + provisioner "local-exec" { + command = <<-EOT + sleep 30 + kubectl -n amazon-cloudwatch get pods,svc -o wide || true + kubectl -n otlp-e2e get pods -o wide || true + kubectl -n otlp-e2e logs deployment/otlp-e2e-probe --tail=6 --prefix || true + kubectl -n amazon-cloudwatch logs ds/cloudwatch-agent --tail=100 --prefix || true + EOT + } +} + +# --- Test runner --- + +resource "null_resource" "validator" { + depends_on = [null_resource.senders, null_resource.agent_diagnostics] + + triggers = { always_run = timestamp() } + + provisioner "local-exec" { + command = <<-EOT + echo "Running OTLP endpoint disabled cluster integration tests" + cd ../../../.. + + echo "Waiting 3 minutes, so that telemetry the agent should not export would have arrived..." + sleep 180 + + go test -tags integration -timeout 1h -v ${var.test_dir} \ + -eksClusterName=${aws_eks_cluster.this.name} \ + -computeType=EKS \ + -eksDeploymentStrategy=DAEMON \ + -region=${var.region} + EOT + } +} diff --git a/terraform/eks/daemon/otel-otlp-disabled/providers.tf b/terraform/eks/daemon/otel-otlp-disabled/providers.tf new file mode 100644 index 000000000..a562a7f16 --- /dev/null +++ b/terraform/eks/daemon/otel-otlp-disabled/providers.tf @@ -0,0 +1,28 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +provider "aws" { + region = var.region +} + +provider "kubernetes" { + host = aws_eks_cluster.this.endpoint + cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data) + exec { + api_version = "client.authentication.k8s.io/v1beta1" + command = "aws" + args = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name] + } +} + +provider "helm" { + kubernetes = { + host = aws_eks_cluster.this.endpoint + cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data) + exec = { + api_version = "client.authentication.k8s.io/v1beta1" + command = "aws" + args = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name] + } + } +} diff --git a/terraform/eks/daemon/otel-otlp-disabled/variables.tf b/terraform/eks/daemon/otel-otlp-disabled/variables.tf new file mode 100644 index 000000000..be5cbf6ba --- /dev/null +++ b/terraform/eks/daemon/otel-otlp-disabled/variables.tf @@ -0,0 +1,49 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +variable "region" { + type = string + default = "us-west-2" +} + +variable "test_dir" { + type = string + default = "./test/otel/otlp_disabled" +} + +variable "cwagent_image_repo" { + type = string + default = "public.ecr.aws/cloudwatch-agent/cloudwatch-agent" +} + +variable "cwagent_image_tag" { + type = string + default = "latest" +} + +# The chart is cloned from helm_chart_repo at helm_chart_branch, so a branch on +# a fork can be tested before it is merged. +variable "helm_chart_repo" { + type = string + default = "https://github.com/aws-observability/helm-charts.git" +} + +variable "helm_chart_branch" { + type = string + default = "main" +} + +variable "k8s_version" { + type = string + default = "1.35" +} + +variable "ami_type" { + type = string + default = "AL2023_x86_64_STANDARD" +} + +variable "instance_type" { + type = string + default = "t3.medium" +} diff --git a/terraform/eks/daemon/otel-otlp/main.tf b/terraform/eks/daemon/otel-otlp/main.tf new file mode 100644 index 000000000..962d2ed8d --- /dev/null +++ b/terraform/eks/daemon/otel-otlp/main.tf @@ -0,0 +1,522 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT + +# The chart's OTLP endpoint with its default values, next to OTel Container +# Insights. Ordinary pods (no hostNetwork) send OTLP through the +# cloudwatch-agent Service and set no pod attributes, so the node agent has to +# find each sender from the connection's source IP. One sender per service +# identity case: see local.senders and test/otel/otlp. + +module "common" { + source = "../../../common" + cwagent_image_repo = var.cwagent_image_repo + cwagent_image_tag = var.cwagent_image_tag +} + +module "basic_components" { + source = "../../../basic_components" + region = var.region +} + +locals { + aws_eks = "aws eks --region ${var.region}" + + # A node label fixture, so the test can check that node labels reach OTLP + # telemetry. kubernetes.io/os and kubernetes.io/arch come from the kubelet. + node_labels = { + "otlp-e2e-node" = "v" + } + + sender_script = "${path.module}/../../../otlp_service_sender.sh" +} + +resource "aws_eks_cluster" "this" { + name = "cwagent-eks-integ-${module.common.testing_id}" + role_arn = module.basic_components.role_arn + version = var.k8s_version + vpc_config { + subnet_ids = module.basic_components.public_subnet_ids + security_group_ids = [module.basic_components.security_group] + } +} + +resource "aws_eks_node_group" "this" { + cluster_name = aws_eks_cluster.this.name + node_group_name = "cwagent-otlp-${module.common.testing_id}" + node_role_arn = aws_iam_role.node_role.arn + subnet_ids = module.basic_components.public_subnet_ids + + # Two nodes, so that a sender on each node proves each node agent resolves + # its own pods. + scaling_config { + desired_size = 2 + max_size = 2 + min_size = 2 + } + + ami_type = var.ami_type + capacity_type = "ON_DEMAND" + disk_size = 20 + instance_types = [var.instance_type] + labels = local.node_labels + + depends_on = [ + aws_iam_role_policy_attachment.node_AmazonEC2ContainerRegistryReadOnly, + aws_iam_role_policy_attachment.node_AmazonEKS_CNI_Policy, + aws_iam_role_policy_attachment.node_AmazonEKSWorkerNodePolicy, + ] +} + +# EKS Node IAM Role +resource "aws_iam_role" "node_role" { + name = "cwagent-otlp-eks-Worker-Role-${module.common.testing_id}" + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [{ + Effect = "Allow" + Principal = { Service = "ec2.amazonaws.com" } + Action = "sts:AssumeRole" + }] + }) +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEKSWorkerNodePolicy" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEKSWorkerNodePolicy" + role = aws_iam_role.node_role.name +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEKS_CNI_Policy" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEKS_CNI_Policy" + role = aws_iam_role.node_role.name +} + +resource "aws_iam_role_policy_attachment" "node_AmazonEC2ContainerRegistryReadOnly" { + policy_arn = "arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly" + role = aws_iam_role.node_role.name +} + +# Pod Identity IAM Role. CloudWatchAgentServerPolicy is the policy customers +# are told to use, and covers the OTLP metrics, logs and traces exports. +resource "aws_iam_role" "pod_identity_role" { + name = "cwagent-otlp-pod-identity-${module.common.testing_id}" + assume_role_policy = jsonencode({ + Version = "2012-10-17" + Statement = [{ + Effect = "Allow" + Principal = { Service = "pods.eks.amazonaws.com" } + Action = ["sts:AssumeRole", "sts:TagSession"] + }] + }) +} + +resource "aws_iam_role_policy_attachment" "pod_identity_CloudWatchAgentServerPolicy" { + policy_arn = "arn:aws:iam::aws:policy/CloudWatchAgentServerPolicy" + role = aws_iam_role.pod_identity_role.name +} + +# --- EKS Addon: Pod Identity agent --- + +resource "aws_eks_addon" "pod_identity_agent" { + depends_on = [aws_eks_node_group.this] + cluster_name = aws_eks_cluster.this.name + addon_name = "eks-pod-identity-agent" +} + +# --- Update kubeconfig --- + +resource "null_resource" "kubectl" { + depends_on = [aws_eks_cluster.this, aws_eks_node_group.this] + provisioner "local-exec" { + command = "${local.aws_eks} update-kubeconfig --name ${aws_eks_cluster.this.name}" + } +} + +# --- Helm chart install --- + +data "external" "clone_helm_chart" { + program = ["bash", "-c", <<-EOT + set -e + rm -rf ./helm-charts + git clone -b ${var.helm_chart_branch} ${var.helm_chart_repo} ./helm-charts >&2 + # A chart without the OTLP endpoint would install fine and fail every + # test; stop here with the reason instead. + if ! grep -qE '^otlp:' ./helm-charts/charts/amazon-cloudwatch-observability/values.yaml; then + echo "${var.helm_chart_repo}@${var.helm_chart_branch} has no top-level otlp block in values.yaml" >&2 + exit 1 + fi + echo '{"status":"ready"}' + EOT + ] +} + +# The chart defaults, so the OTLP endpoint is on with its default ports and +# span metrics, plus OTel Container Insights to check that both run together. +resource "helm_release" "aws_observability" { + name = "amazon-cloudwatch-observability" + chart = "./helm-charts/charts/amazon-cloudwatch-observability" + namespace = "amazon-cloudwatch" + create_namespace = true + + set = [ + { name = "clusterName", value = aws_eks_cluster.this.name }, + { name = "region", value = var.region }, + { name = "otelContainerInsights.enabled", value = "true" }, + ] + + depends_on = [ + aws_eks_addon.pod_identity_agent, + null_resource.kubectl, + data.external.clone_helm_chart, + ] +} + +# --- Pod Identity association (after Helm creates the service account) --- + +resource "aws_eks_pod_identity_association" "cloudwatch_agent" { + depends_on = [helm_release.aws_observability] + cluster_name = aws_eks_cluster.this.name + namespace = "amazon-cloudwatch" + service_account = "cloudwatch-agent" + role_arn = aws_iam_role.pod_identity_role.arn +} + +# --- Patch agent image (both DaemonSet and cluster-scraper) --- + +resource "null_resource" "update_image" { + depends_on = [helm_release.aws_observability, null_resource.kubectl] + triggers = { timestamp = timestamp() } + provisioner "local-exec" { + command = <<-EOT + set -e + sleep 30 + kubectl -n amazon-cloudwatch patch AmazonCloudWatchAgent cloudwatch-agent --type='json' \ + -p='[{"op": "replace", "path": "/spec/image", "value": "${var.cwagent_image_repo}:${var.cwagent_image_tag}"}]' + kubectl -n amazon-cloudwatch patch AmazonCloudWatchAgent cloudwatch-agent-cluster-scraper --type='json' \ + -p='[{"op": "replace", "path": "/spec/image", "value": "${var.cwagent_image_repo}:${var.cwagent_image_tag}"}]' 2>/dev/null || true + sleep 10 + EOT + } +} + +# --- Restart pods to pick up Pod Identity + new image --- + +resource "null_resource" "restart_pods" { + depends_on = [aws_eks_pod_identity_association.cloudwatch_agent, null_resource.update_image] + triggers = { timestamp = timestamp() } + provisioner "local-exec" { + command = <<-EOT + set -e + kubectl -n amazon-cloudwatch rollout restart daemonset/cloudwatch-agent + kubectl -n amazon-cloudwatch rollout restart deployment/cloudwatch-agent-cluster-scraper 2>/dev/null || true + kubectl -n amazon-cloudwatch rollout status daemonset/cloudwatch-agent --timeout=120s + kubectl -n amazon-cloudwatch rollout status deployment/cloudwatch-agent-cluster-scraper --timeout=120s 2>/dev/null || true + EOT + } +} + +# --- OTLP senders, one per service identity case --- +# Every sender runs terraform/otlp_service_sender.sh from a ConfigMap and posts +# to the default endpoint, http://cloudwatch-agent.amazon-cloudwatch:4318. Each +# pod has one container, so the agent also sets k8s.container.name. The +# expected service.name of each case is in test/otel/otlp/otlp_test.go. +# +# app app.kubernetes.io/name and /version labels, no service.name sent; +# two replicas, one per node +# annotation resource.opentelemetry.io/service.name and /service.version +# annotations, which win over the app labels +# explicit the sender sends service.name, which wins over the app label +# unknown the sender sends unknown_service:sh, which counts as not set, +# so the Deployment name is used +# daemonset a DaemonSet with no labels or annotations for identity +# cronjob a CronJob with no labels or annotations for identity +# +# The senders start after the agents restart, so that the first records are +# not sent to an agent that is going away. + +resource "null_resource" "senders" { + depends_on = [null_resource.restart_pods] + provisioner "local-exec" { + command = <<-EOT + set -e + kubectl create namespace otlp-e2e --dry-run=client -o yaml | kubectl apply -f - + kubectl -n otlp-e2e create configmap otlp-e2e-sender \ + --from-file=sender.sh=${local.sender_script} --dry-run=client -o yaml | kubectl apply -f - + cat <<'EOF' | kubectl apply -f - + apiVersion: apps/v1 + kind: Deployment + metadata: + name: otlp-e2e-app + namespace: otlp-e2e + spec: + replicas: 2 + selector: + matchLabels: + app: otlp-e2e-app + template: + metadata: + labels: + app: otlp-e2e-app + app.kubernetes.io/name: otlp-e2e-app-name + app.kubernetes.io/version: "1.2.3" + spec: + affinity: + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - labelSelector: + matchLabels: + app: otlp-e2e-app + topologyKey: kubernetes.io/hostname + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: app + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + --- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: otlp-e2e-annotated + namespace: otlp-e2e + spec: + replicas: 1 + selector: + matchLabels: + app: otlp-e2e-annotated + template: + metadata: + labels: + app: otlp-e2e-annotated + app.kubernetes.io/name: otlp-e2e-label-loses + app.kubernetes.io/version: "0.0.1" + annotations: + resource.opentelemetry.io/service.name: otlp-e2e-annotated-svc + resource.opentelemetry.io/service.version: "9.9.9" + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: annotation + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + --- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: otlp-e2e-explicit + namespace: otlp-e2e + spec: + replicas: 1 + selector: + matchLabels: + app: otlp-e2e-explicit + template: + metadata: + labels: + app: otlp-e2e-explicit + app.kubernetes.io/name: otlp-e2e-label-loses + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: explicit + - name: OTLP_SERVICE_NAME + value: otlp-e2e-explicit-svc + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + --- + apiVersion: apps/v1 + kind: Deployment + metadata: + name: otlp-e2e-unknown + namespace: otlp-e2e + spec: + replicas: 1 + selector: + matchLabels: + app: otlp-e2e-unknown + template: + metadata: + labels: + app: otlp-e2e-unknown + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: unknown + - name: OTLP_SERVICE_NAME + value: unknown_service:sh + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + --- + apiVersion: apps/v1 + kind: DaemonSet + metadata: + name: otlp-e2e-ds + namespace: otlp-e2e + spec: + selector: + matchLabels: + app: otlp-e2e-ds + template: + metadata: + labels: + app: otlp-e2e-ds + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: daemonset + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + --- + apiVersion: batch/v1 + kind: CronJob + metadata: + name: otlp-e2e-cron + namespace: otlp-e2e + spec: + schedule: "* * * * *" + concurrencyPolicy: Forbid + successfulJobsHistoryLimit: 1 + failedJobsHistoryLimit: 1 + jobTemplate: + spec: + backoffLimit: 0 + template: + metadata: + labels: + app: otlp-e2e-cron + spec: + restartPolicy: Never + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: cronjob + # Three rounds 10s apart, so a round arrives after the + # agent's pod cache has the new pod's IP. + - name: OTLP_ITERATIONS + value: "3" + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender + EOF + for w in deployment/otlp-e2e-app deployment/otlp-e2e-annotated deployment/otlp-e2e-explicit \ + deployment/otlp-e2e-unknown daemonset/otlp-e2e-ds; do + kubectl -n otlp-e2e rollout status "$w" --timeout=5m + done + EOT + } +} + +# --- Diagnostics: agent and sender state in the job output, so a failed run +# can be debugged after the cluster is destroyed --- + +resource "null_resource" "agent_diagnostics" { + depends_on = [null_resource.senders] + triggers = { always_run = timestamp() } + provisioner "local-exec" { + command = <<-EOT + sleep 30 + kubectl -n amazon-cloudwatch get pods,svc -o wide || true + kubectl -n otlp-e2e get pods -o wide || true + kubectl -n otlp-e2e logs deployment/otlp-e2e-app --tail=6 --prefix || true + kubectl -n amazon-cloudwatch logs ds/cloudwatch-agent --tail=100 --prefix || true + EOT + } +} + +# --- Test runner --- + +resource "null_resource" "validator" { + depends_on = [null_resource.senders, null_resource.agent_diagnostics] + + triggers = { always_run = timestamp() } + + provisioner "local-exec" { + command = <<-EOT + echo "Running OTLP endpoint cluster integration tests" + cd ../../../.. + + echo "Waiting 3 minutes for metrics, logs and traces to propagate..." + sleep 180 + + go test -tags integration -timeout 1h -v ${var.test_dir} \ + -eksClusterName=${aws_eks_cluster.this.name} \ + -computeType=EKS \ + -eksDeploymentStrategy=DAEMON \ + -region=${var.region} + EOT + } +} diff --git a/terraform/eks/daemon/otel-otlp/providers.tf b/terraform/eks/daemon/otel-otlp/providers.tf new file mode 100644 index 000000000..a562a7f16 --- /dev/null +++ b/terraform/eks/daemon/otel-otlp/providers.tf @@ -0,0 +1,28 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +provider "aws" { + region = var.region +} + +provider "kubernetes" { + host = aws_eks_cluster.this.endpoint + cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data) + exec { + api_version = "client.authentication.k8s.io/v1beta1" + command = "aws" + args = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name] + } +} + +provider "helm" { + kubernetes = { + host = aws_eks_cluster.this.endpoint + cluster_ca_certificate = base64decode(aws_eks_cluster.this.certificate_authority.0.data) + exec = { + api_version = "client.authentication.k8s.io/v1beta1" + command = "aws" + args = ["eks", "get-token", "--cluster-name", aws_eks_cluster.this.name] + } + } +} diff --git a/terraform/eks/daemon/otel-otlp/variables.tf b/terraform/eks/daemon/otel-otlp/variables.tf new file mode 100644 index 000000000..daa180f2f --- /dev/null +++ b/terraform/eks/daemon/otel-otlp/variables.tf @@ -0,0 +1,49 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +variable "region" { + type = string + default = "us-west-2" +} + +variable "test_dir" { + type = string + default = "./test/otel/otlp" +} + +variable "cwagent_image_repo" { + type = string + default = "public.ecr.aws/cloudwatch-agent/cloudwatch-agent" +} + +variable "cwagent_image_tag" { + type = string + default = "latest" +} + +# The chart is cloned from helm_chart_repo at helm_chart_branch, so a branch on +# a fork can be tested before it is merged. +variable "helm_chart_repo" { + type = string + default = "https://github.com/aws-observability/helm-charts.git" +} + +variable "helm_chart_branch" { + type = string + default = "main" +} + +variable "k8s_version" { + type = string + default = "1.35" +} + +variable "ami_type" { + type = string + default = "AL2023_x86_64_STANDARD" +} + +variable "instance_type" { + type = string + default = "t3.medium" +} diff --git a/terraform/otlp_service_sender.sh b/terraform/otlp_service_sender.sh new file mode 100644 index 000000000..f51c2c0cf --- /dev/null +++ b/terraform/otlp_service_sender.sh @@ -0,0 +1,68 @@ +#!/bin/sh +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT +# +# OTLP sender for the Kubernetes tests of the chart's OTLP endpoint. It runs in an ordinary pod (no +# hostNetwork) and posts OTLP/HTTP JSON to the cloudwatch-agent Service, the way an application does. +# It sets no pod or node attributes, so the agent has to find the pod from the connection's source IP. +# +# Unlike otlp_load_generator.sh, this file is not a Terraform template: it is mounted from a ConfigMap +# as is, so shell expansions are written normally. +# +# Environment: +# OTLP_ENDPOINT base URL of the receiver (default http://cloudwatch-agent.amazon-cloudwatch:4318) +# OTLP_CASE value of the otlp_e2e_case attribute on every metric data point, log record and span +# OTLP_SERVICE_NAME service.name resource attribute to send; empty sends none +# OTLP_METRICS space-separated names of the counters to send (default otlp_e2e_requests) +# OTLP_ITERATIONS rounds to send; 0 sends until the pod stops (default 0) +# OTLP_INTERVAL seconds between rounds (default 10) +# +# Every post prints "otlp-e2e-send " to stdout, with status 000 when the +# connection fails, so a test can read from the pod log whether the endpoint answered. + +ENDPOINT="${OTLP_ENDPOINT:-http://cloudwatch-agent.amazon-cloudwatch:4318}" +CASE="${OTLP_CASE:-default}" +METRICS="${OTLP_METRICS:-otlp_e2e_requests}" +ITERATIONS="${OTLP_ITERATIONS:-0}" +INTERVAL="${OTLP_INTERVAL:-10}" + +if [ -n "$OTLP_SERVICE_NAME" ]; then + RESOURCE="[{\"key\":\"service.name\",\"value\":{\"stringValue\":\"$OTLP_SERVICE_NAME\"}}]" +else + RESOURCE="[]" +fi +CASE_ATTR="{\"key\":\"otlp_e2e_case\",\"value\":{\"stringValue\":\"$CASE\"}}" +START_NS="$(date +%s)000000000" + +hex() { + od -An -N"$1" -tx1 /dev/urandom | tr -d ' \n' +} + +post() { + code=$(curl -s -o /dev/null -w '%{http_code}' --max-time 5 -X POST "$ENDPOINT/v1/$1" \ + -H 'Content-Type: application/json' -d "$2") + echo "otlp-e2e-send $1 $code" +} + +SEQ=0 +while [ "$ITERATIONS" -eq 0 ] || [ "$SEQ" -lt "$ITERATIONS" ]; do + SEQ=$((SEQ + 1)) + NOW_S=$(date +%s) + NOW_NS="${NOW_S}000000000" + + POINTS="" + for name in $METRICS; do + [ -n "$POINTS" ] && POINTS="$POINTS," + POINTS="$POINTS{\"name\":\"$name\",\"unit\":\"1\",\"sum\":{\"aggregationTemporality\":2,\"isMonotonic\":true,\"dataPoints\":[{\"asInt\":\"$SEQ\",\"startTimeUnixNano\":\"$START_NS\",\"timeUnixNano\":\"$NOW_NS\",\"attributes\":[$CASE_ATTR]}]}}" + done + post metrics "{\"resourceMetrics\":[{\"resource\":{\"attributes\":$RESOURCE},\"scopeMetrics\":[{\"scope\":{\"name\":\"otlp-e2e\"},\"metrics\":[$POINTS]}]}]}" + + post logs "{\"resourceLogs\":[{\"resource\":{\"attributes\":$RESOURCE},\"scopeLogs\":[{\"scope\":{\"name\":\"otlp-e2e\"},\"logRecords\":[{\"timeUnixNano\":\"$NOW_NS\",\"severityText\":\"INFO\",\"body\":{\"stringValue\":\"otlp-e2e $CASE\"},\"attributes\":[$CASE_ATTR]}]}]}]}" + + # X-Ray needs the first 4 bytes of the trace ID to be the Unix time in seconds. The span is a + # server span with HTTP attributes, so the span metrics connector has dimensions to report. + TRACE_ID="$(printf '%08x' "$NOW_S")$(hex 12)" + post traces "{\"resourceSpans\":[{\"resource\":{\"attributes\":$RESOURCE},\"scopeSpans\":[{\"scope\":{\"name\":\"otlp-e2e\"},\"spans\":[{\"traceId\":\"$TRACE_ID\",\"spanId\":\"$(hex 8)\",\"name\":\"GET /e2e\",\"kind\":2,\"startTimeUnixNano\":\"$((NOW_S - 1))000000000\",\"endTimeUnixNano\":\"$NOW_NS\",\"attributes\":[$CASE_ATTR,{\"key\":\"http.request.method\",\"value\":{\"stringValue\":\"GET\"}},{\"key\":\"http.route\",\"value\":{\"stringValue\":\"/e2e\"}},{\"key\":\"http.response.status_code\",\"value\":{\"intValue\":\"200\"}}]}]}]}]}" + + sleep "$INTERVAL" +done diff --git a/test/azure/aks/containerinsights/aks_otlp_test.go b/test/azure/aks/containerinsights/aks_otlp_test.go new file mode 100644 index 000000000..6b81f0557 --- /dev/null +++ b/test/azure/aks/containerinsights/aks_otlp_test.go @@ -0,0 +1,225 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +//go:build integration + +package containerinsights + +import ( + "encoding/json" + "fmt" + "log" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" + "github.com/aws/amazon-cloudwatch-agent-test/test/otel_collect/otlpvalidation" + "github.com/aws/amazon-cloudwatch-agent-test/test/status" + "github.com/aws/amazon-cloudwatch-agent-test/util/awsservice" +) + +// The OTLP sender in resources/otlp_sender.yaml: an ordinary pod that sends through the +// cloudwatch-agent Service to the chart's OTLP endpoint, with the app.kubernetes.io/name and version +// labels and no service.name. +const ( + otlpNamespace = "otlp-e2e" + otlpService = "otlp-e2e-app-name" + otlpCase = "app" +) + +// otlpMetricNames is every metric the chart's OTLP pipelines export for this cluster: the sender's +// counter and the span metrics computed from its spans. +var otlpMetricNames = []string{otlpendpoint.MetricName, "traces.span.metrics.calls", "traces.span.metrics.duration"} + +// otlpResourceExpectations are the resource attributes of the sender's telemetry: the AKS resource +// detection of the node, the Container Insights enrichment of the pod, its labels and the service +// identity. Node-level and subscription-bearing values are presence-only, as in the default:otel +// suite. cloud.resource_id carries the subscription and is checked separately, without logging it. +func otlpResourceExpectations() map[string]string { + return map[string]string{ + "cloud.provider": "azure", + "cloud.platform": "azure.aks", + "cloud.region": env.AzureLocation, + "azure.vm.size": env.AzureVMSize, + "azure.resourcegroup.name": env.AzureResourceGroup, + "k8s.cluster.name": env.AKSClusterName, + "k8s.namespace.name": otlpNamespace, + "k8s.deployment.name": "otlp-e2e-app", + "k8s.workload.name": "otlp-e2e-app", + "k8s.workload.type": "Deployment", + "k8s.container.name": "sender", + "k8s.pod.label.app": "otlp-e2e-app", + "k8s.pod.label.app.kubernetes.io/name": otlpService, + "k8s.node.label.kubernetes.io/os": "linux", + "service.name": otlpService, + "service.namespace": otlpNamespace, + "service.version": "1.2.3", + "deployment.environment.name": "azure.aks:" + env.AKSClusterName + "/" + otlpNamespace, + "cloud.account.id": otlpvalidation.PresenceOnly, + "cloud.resource_id": otlpvalidation.PresenceOnly, + "k8s.pod.name": otlpvalidation.PresenceOnly, + "k8s.pod.uid": otlpvalidation.PresenceOnly, + "k8s.node.name": otlpvalidation.PresenceOnly, + "service.instance.id": otlpvalidation.PresenceOnly, + "host.id": otlpvalidation.PresenceOnly, + "host.name": otlpvalidation.PresenceOnly, + "azure.vm.name": otlpvalidation.PresenceOnly, + "azure.vm.scaleset.name": otlpvalidation.PresenceOnly, + } +} + +// TestAKSOTLP validates the chart's OTLP endpoint, which the chart serves by default, with the install +// values the AKS setup script uses (default:otel is an alias of the default config). The sender's +// metrics, logs and traces must carry the Container Insights enrichment of its pod. +func TestAKSOTLP(t *testing.T) { + resourceLabels := map[string]string{} + for attr, want := range otlpResourceExpectations() { + resourceLabels["@resource."+attr] = otlpvalidation.ExpectedValue(want) + } + scopeLabels := map[string]string{ + "@instrumentation.cloudwatch.source": otlpvalidation.ExactMatch(otlpvalidation.CloudWatchSource), + "@instrumentation.cloudwatch.solution": otlpvalidation.ExactMatch(otlpendpoint.Solution), + } + withScope := func(extra map[string]string) map[string]string { + labels := map[string]string{} + for _, m := range []map[string]string{resourceLabels, scopeLabels, extra} { + for k, v := range m { + labels[k] = v + } + } + return labels + } + assertFound := func(t *testing.T, group status.TestGroupResult) { + require.NotEmpty(t, group.TestResults) + for _, r := range group.TestResults { + require.Equal(t, status.SUCCESSFUL, r.Status, "metric %s: %v", r.Name, r.Reason) + } + } + + t.Run("Metrics", func(t *testing.T) { + assertFound(t, otlpvalidation.ValidateOtlpMetricsWithLabels("AKSOTLPMetric", env.Region, + []string{otlpendpoint.MetricName}, + withScope(map[string]string{otlpendpoint.CaseAttribute: otlpvalidation.ExactMatch(otlpCase)}))) + }) + t.Run("SpanMetrics", func(t *testing.T) { + assertFound(t, otlpvalidation.ValidateOtlpMetricsWithLabels("AKSOTLPSpanMetrics", env.Region, + otlpMetricNames[1:], + withScope(map[string]string{"http.request.method": otlpvalidation.ExactMatch("GET")}))) + }) + t.Run("Names", func(t *testing.T) { + matcher := fmt.Sprintf(`{__name__=~".+","@resource.k8s.cluster.name"=%q,"@instrumentation.cloudwatch.solution"=%q}`, + env.AKSClusterName, otlpendpoint.Solution) + got, err := otlpvalidation.MetricNamesForMatcher(env.Region, matcher, time.Now().Add(-time.Hour), time.Now()) + require.NoError(t, err) + require.ElementsMatch(t, otlpMetricNames, got) + }) + t.Run("Logs", func(t *testing.T) { + r := validateOTLPLogs() + require.Equal(t, status.SUCCESSFUL, r.Status, "OTLP log validation failed: %v", r.Reason) + }) + t.Run("Traces", func(t *testing.T) { + r := validateOTLPTraces() + require.Equal(t, status.SUCCESSFUL, r.Status, "OTLP trace validation failed: %v", r.Reason) + }) +} + +// assertResourceID checks cloud.resource_id without putting the subscription in the error. +func assertResourceID(scope string, attrs map[string]any) error { + want := "/providers/Microsoft.ContainerService/managedClusters/" + env.AKSClusterName + got := fmt.Sprint(attrs["cloud.resource_id"]) + if !strings.HasPrefix(got, "/subscriptions/") || !strings.HasSuffix(got, want) { + return fmt.Errorf("%s: cloud.resource_id is not the AKS cluster's resource ID (want /subscriptions/...%s)", scope, want) + } + return nil +} + +func validateOTLPLogs() status.TestResult { + testResult := status.TestResult{Name: "AKS_OTLP_Logs", Status: status.FAILED} + logGroup := otlpendpoint.LogGroup(env.AKSClusterName) + defer func() { + if testResult.Status == status.SUCCESSFUL { + awsservice.DeleteLogGroup(logGroup) + } + }() + // {k8s.namespace.name}/{service.namespace}/{service.name}, with the identity the chart derives. + logStream := strings.Join([]string{otlpNamespace, otlpNamespace, otlpService}, "/") + const maxRetries = 4 + const retryInterval = 30 * time.Second + for attempt := 1; attempt <= maxRetries; attempt++ { + since := time.Now().Add(-validationWindow) + until := time.Now() + err := awsservice.ValidateLogs( + logGroup, logStream, &since, &until, + awsservice.AssertLogsNotEmpty(), + awsservice.AssertPerLog(otlpvalidation.AssertLogRecord(func(rec otlpvalidation.LogRecord) error { + if err := otlpvalidation.AssertLogContent(rec, otlpendpoint.LogBody(otlpCase), "INFO", otlpResourceExpectations()); err != nil { + return err + } + return assertResourceID("log", rec.Resource.Attributes) + })), + ) + if err == nil { + testResult.Status = status.SUCCESSFUL + return testResult + } + testResult.Reason = err + if attempt < maxRetries { + log.Printf("[AKS_OTLP_Logs] %v, retrying in %v", err, retryInterval) + time.Sleep(retryInterval) + } + } + return testResult +} + +func validateOTLPTraces() status.TestResult { + testResult := status.TestResult{Name: "AKS_OTLP_Traces", Status: status.FAILED} + query := fmt.Sprintf(`fields @message | filter @message like %q and @message like %q | limit 100`, + otlpService, env.AKSClusterName) + const maxRetries = 5 + const retryInterval = 60 * time.Second + for attempt := 1; attempt <= maxRetries; attempt++ { + since := time.Now().Add(-validationWindow) + rows, err := awsservice.GetLogQueryResults("aws/spans", since.Unix(), time.Now().Unix(), query) + if err != nil { + testResult.Reason = fmt.Errorf("attempt %d: aws/spans query failed: %w", attempt, err) + } else { + found := 0 + for _, row := range rows { + for _, field := range row { + if aws.ToString(field.Field) != "@message" { + continue + } + var s otlpvalidation.SpanRecord + if json.Unmarshal([]byte(aws.ToString(field.Value)), &s) != nil { + continue + } + // Content is stable, so a mismatch is final. + attrs := map[string]string{otlpendpoint.CaseAttribute: otlpCase, "http.request.method": "GET"} + if cerr := otlpvalidation.AssertSpanContent(s, "GET /e2e", "SERVER", attrs, otlpResourceExpectations()); cerr != nil { + testResult.Reason = cerr + return testResult + } + if cerr := assertResourceID("span "+s.TraceID, s.Resource.Attributes); cerr != nil { + testResult.Reason = cerr + return testResult + } + found++ + } + } + if found > 0 { + testResult.Status = status.SUCCESSFUL + return testResult + } + testResult.Reason = fmt.Errorf("attempt %d: no spans of %s in aws/spans", attempt, otlpService) + } + if attempt < maxRetries { + log.Printf("[AKS_OTLP_Traces] %v, retrying in %v", testResult.Reason, retryInterval) + time.Sleep(retryInterval) + } + } + return testResult +} diff --git a/test/azure/aks/containerinsights/resources/otlp_sender.yaml b/test/azure/aks/containerinsights/resources/otlp_sender.yaml new file mode 100644 index 000000000..93743a8ff --- /dev/null +++ b/test/azure/aks/containerinsights/resources/otlp_sender.yaml @@ -0,0 +1,41 @@ +# An ordinary pod (no hostNetwork) that sends OTLP through the cloudwatch-agent +# Service, the chart's OTLP endpoint, which the chart serves by default. It runs +# terraform/otlp_service_sender.sh, which Terraform puts in the otlp-e2e-sender +# ConfigMap, and sets no pod attributes, so the agent has to find the pod from +# the connection's source IP. The labels give the service identity the test +# expects (service.name otlp-e2e-app-name, service.version 1.2.3). +apiVersion: apps/v1 +kind: Deployment +metadata: + name: otlp-e2e-app + namespace: otlp-e2e +spec: + replicas: 1 + selector: + matchLabels: + app: otlp-e2e-app + template: + metadata: + labels: + app: otlp-e2e-app + app.kubernetes.io/name: otlp-e2e-app-name + app.kubernetes.io/version: "1.2.3" + spec: + containers: + - name: sender + image: curlimages/curl:8.8.0 + command: ["sh", "/sender/sender.sh"] + env: + - name: OTLP_CASE + value: app + resources: + requests: + cpu: 10m + memory: 16Mi + volumeMounts: + - name: sender + mountPath: /sender + volumes: + - name: sender + configMap: + name: otlp-e2e-sender diff --git a/test/otel/filters/filters_test.go b/test/otel/filters/filters_test.go index 8ed556578..0b829309e 100644 --- a/test/otel/filters/filters_test.go +++ b/test/otel/filters/filters_test.go @@ -393,26 +393,43 @@ func (r logRecord) resource() map[string]string { // ("application") or of the host log writer ("host") from the last // logLookback. func markerLogs(kind string) ([]logRecord, error) { - logGroup := fmt.Sprintf("/aws/otel/containerinsights/%s/%s", cfg.ClusterName, kind) - end := time.Now() - rows, err := awsservice.GetLogQueryResults(logGroup, end.Add(-logLookback).Unix(), end.Unix(), - fmt.Sprintf("fields @message | filter @message like %q | limit 10000", logMarker)) + return logsIn(fmt.Sprintf("/aws/otel/containerinsights/%s/%s", cfg.ClusterName, kind), logMarker) +} + +// logsIn returns the log records in logGroup from the last logLookback whose +// body starts with marker. +func logsIn(logGroup, marker string) ([]logRecord, error) { + messages, err := queryMessages(logGroup, fmt.Sprintf("fields @message | filter @message like %q | limit 10000", marker)) if err != nil { return nil, err } var records []logRecord + for _, m := range messages { + var r logRecord + if json.Unmarshal([]byte(m), &r) == nil && strings.HasPrefix(r.Body, marker) { + records = append(records, r) + } + } + return records, nil +} + +// queryMessages runs a Logs Insights query over logGroup for the last +// logLookback and returns the @message of each row. +func queryMessages(logGroup, query string) ([]string, error) { + end := time.Now() + rows, err := awsservice.GetLogQueryResults(logGroup, end.Add(-logLookback).Unix(), end.Unix(), query) + if err != nil { + return nil, err + } + var messages []string for _, row := range rows { for _, field := range row { - if aws.ToString(field.Field) != "@message" { - continue - } - var r logRecord - if json.Unmarshal([]byte(aws.ToString(field.Value)), &r) == nil && strings.HasPrefix(r.Body, logMarker) { - records = append(records, r) + if aws.ToString(field.Field) == "@message" { + messages = append(messages, aws.ToString(field.Value)) } } } - return records, nil + return messages, nil } // fromEveryNode reports whether records include one from every node. diff --git a/test/otel/filters/otlp_test.go b/test/otel/filters/otlp_test.go new file mode 100644 index 000000000..07597ad79 --- /dev/null +++ b/test/otel/filters/otlp_test.go @@ -0,0 +1,172 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package filters + +import ( + "context" + "encoding/json" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" + "github.com/aws/amazon-cloudwatch-agent-test/test/otel_collect/otlpvalidation" + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// The OTLP sender container in every test workload (main.tf, workloads) sends otlpendpoint.MetricName, +// which metricNames.include names, and otlpUnlistedMetric, which it does not, plus a log record and a +// server span, through the cloudwatch-agent Service. The chart's OTLP pipelines apply the same +// otelFilters as Container Insights, so these tests reuse the namespace and label lists above. + +const ( + otlpCase = "filters" + otlpUnlistedMetric = "otlp_e2e_unlisted" +) + +// otlpSpanMetrics are computed from the sender's spans, and metricNames.include does not name them. +var otlpSpanMetrics = []string{"traces.span.metrics.calls", "traces.span.metrics.duration"} + +// TestOTLPNamespaceFilter checks that the namespaces filter drops OTLP metrics, logs and spans from +// the dropped namespaces. Absence is checked only after the signal is seen for every kept namespace, +// from every node for metrics and logs. +func TestOTLPNamespaceFilter(t *testing.T) { + for _, ns := range keptNamespaces { + seen(t, otlpendpoint.MetricName+" in "+ns, otlpendpoint.MetricName, map[string]string{"@resource.k8s.namespace.name": ns}) + } + for _, ns := range droppedNamespaces { + results, err := query(context.Background(), otlpendpoint.MetricName, map[string]string{"@resource.k8s.namespace.name": ns}) + require.NoError(t, err) + assert.Empty(t, results, "OTLP metric present for filtered namespace %s", ns) + } + + records := eventually(t, "OTLP logs from included namespaces on every node", func() ([]logRecord, error) { + records, err := otlpLogs() + if err != nil { + return nil, err + } + for _, ns := range keptNamespaces { + if !fromEveryNode(byNamespace(records, ns)) { + return nil, nil + } + } + return records, nil + }) + for _, ns := range droppedNamespaces { + assert.Empty(t, byNamespace(records, ns), "OTLP logs present for filtered namespace %s", ns) + } + + spans := eventually(t, "OTLP spans from included namespaces", func() ([]otlpvalidation.SpanRecord, error) { + spans, err := otlpSpans() + if err != nil { + return nil, err + } + for _, ns := range keptNamespaces { + if len(spansIn(spans, ns)) == 0 { + return nil, nil + } + } + return spans, nil + }) + for _, ns := range droppedNamespaces { + assert.Empty(t, spansIn(spans, ns), "OTLP spans present for filtered namespace %s", ns) + } +} + +// TestOTLPLabelFilter checks the pod and node label filters on OTLP metrics, logs and spans. Each +// signal has its own label processors in the chart's OTLP pipelines. +func TestOTLPLabelFilter(t *testing.T) { + inKeepA := map[string]string{"@resource.k8s.namespace.name": "filters-keep-a"} + + const metricWhat = "OTLP metric in filters-keep-a" + results := eventually(t, metricWhat, func() ([]otelmetrics.MetricResult, error) { + return query(context.Background(), otlpendpoint.MetricName, inKeepA) + }) + resources := make([]map[string]string, 0, len(results)) + for _, r := range results { + resources = append(resources, r.Labels.Resource) + } + assertLabels(t, metricWhat, resources, keptLabels, droppedLabels) + + const logWhat = "OTLP logs from filters-keep-a" + records := eventually(t, logWhat, func() ([]logRecord, error) { + records, err := otlpLogs() + return byNamespace(records, "filters-keep-a"), err + }) + resources = resources[:0] + for _, r := range records { + resources = append(resources, r.resource()) + } + assertLabels(t, logWhat, resources, keptLabels, droppedLabels) + + const spanWhat = "OTLP spans from filters-keep-a" + spans := eventually(t, spanWhat, func() ([]otlpvalidation.SpanRecord, error) { + spans, err := otlpSpans() + return spansIn(spans, "filters-keep-a"), err + }) + resources = resources[:0] + for _, s := range spans { + resources = append(resources, stringMap(s.Resource.Attributes)) + } + assertLabels(t, spanWhat, resources, keptLabels, droppedLabels) +} + +// TestOTLPMetricNameFilter checks that a non-empty metricNames.include drops every OTLP metric it does +// not name, span metrics included. Absence is checked only after the named metric is seen from every +// node in an included namespace, so the senders and the pipeline are known to be reporting. +func TestOTLPMetricNameFilter(t *testing.T) { + const ns = "filters-keep-a" + inNamespace := map[string]string{"@resource.k8s.namespace.name": ns} + seen(t, otlpendpoint.MetricName+" in "+ns, otlpendpoint.MetricName, inNamespace) + for _, metric := range append([]string{otlpUnlistedMetric}, otlpSpanMetrics...) { + results, err := query(context.Background(), metric, inNamespace) + require.NoError(t, err) + assert.Empty(t, results, "filtered OTLP metric %s present in %s", metric, ns) + } +} + +// otlpLogs returns the sender's OTLP log records from the last logLookback. +func otlpLogs() ([]logRecord, error) { + return logsIn(otlpendpoint.LogGroup(cfg.ClusterName), otlpendpoint.LogBody(otlpCase)) +} + +// otlpSpans returns the sender's spans of this cluster from Transaction Search, from the last +// logLookback. +func otlpSpans() ([]otlpvalidation.SpanRecord, error) { + messages, err := queryMessages("aws/spans", fmt.Sprintf( + "fields @message | filter @message like %q and @message like %q | limit 10000", cfg.ClusterName, otlpendpoint.CaseAttribute)) + if err != nil { + return nil, err + } + var spans []otlpvalidation.SpanRecord + for _, m := range messages { + var s otlpvalidation.SpanRecord + if json.Unmarshal([]byte(m), &s) == nil && fmt.Sprint(s.Attributes[otlpendpoint.CaseAttribute]) == otlpCase { + spans = append(spans, s) + } + } + return spans, nil +} + +func spansIn(spans []otlpvalidation.SpanRecord, ns string) []otlpvalidation.SpanRecord { + var out []otlpvalidation.SpanRecord + for _, s := range spans { + if fmt.Sprint(s.Resource.Attributes["k8s.namespace.name"]) == ns { + out = append(out, s) + } + } + return out +} + +func stringMap(m map[string]any) map[string]string { + out := make(map[string]string, len(m)) + for k, v := range m { + out[k] = fmt.Sprint(v) + } + return out +} diff --git a/test/otel/otlp/otlp_test.go b/test/otel/otlp/otlp_test.go new file mode 100644 index 000000000..feff56d2e --- /dev/null +++ b/test/otel/otlp/otlp_test.go @@ -0,0 +1,541 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +// Package otlp tests the Helm chart's OTLP endpoint on EKS, installed with the chart defaults next to +// OTel Container Insights (terraform/eks/daemon/otel-otlp). Ordinary pods send OTLP through the +// cloudwatch-agent Service without pod attributes, and the tests check what reaches CloudWatch: the +// Container Insights enrichment of the sender pod, labels included, the service identity, span metrics, +// the logs routing, traces, and that Container Insights itself is unchanged. +package otlp + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" + "github.com/aws/amazon-cloudwatch-agent-test/test/otel_collect/otlpvalidation" + "github.com/aws/amazon-cloudwatch-agent-test/util/awsservice" + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// The sender workloads in terraform/eks/daemon/otel-otlp/main.tf. +const ( + senderNamespace = "otlp-e2e" + senderContainer = "sender" + + // The app case: a Deployment with two replicas, one per node, and the app.kubernetes.io/name and + // version labels. + appDeployment = "otlp-e2e-app" + appService = "otlp-e2e-app-name" + appVersion = "1.2.3" + + // containerInsightsSolution is the cloudwatch.solution of OTel Container Insights telemetry. + containerInsightsSolution = "k8s-otel-container-insights" +) + +const ( + lookback = 15 * time.Minute + attempts = 10 + retryInterval = 30 * time.Second +) + +// identityCases are the service identity cases, one sender workload each. Every case checks +// service.name, and some check the other attributes that tell how the agent identified the sender. +var identityCases = []struct { + senderCase string + // everyNode means the case has a sender on every node, so each node agent must resolve it. + everyNode bool + want map[string]string +}{ + // The app.kubernetes.io/name and version labels, with no service.name sent. + {"app", true, map[string]string{ + "service.name": appService, + "service.version": appVersion, + }}, + // The resource.opentelemetry.io/service.* annotations win over the app labels. + {"annotation", false, map[string]string{ + "service.name": "otlp-e2e-annotated-svc", + "service.version": "9.9.9", + }}, + // A service.name the sender set wins over the app label. + {"explicit", false, map[string]string{ + "service.name": "otlp-e2e-explicit-svc", + }}, + // unknown_service:sh counts as not set, so the Deployment name is used. + {"unknown", false, map[string]string{ + "service.name": "otlp-e2e-unknown", + "k8s.deployment.name": "otlp-e2e-unknown", + }}, + // A DaemonSet pod with no identity labels: the DaemonSet name. + {"daemonset", true, map[string]string{ + "service.name": "otlp-e2e-ds", + "k8s.daemonset.name": "otlp-e2e-ds", + "k8s.workload.name": "otlp-e2e-ds", + "k8s.workload.type": "DaemonSet", + }}, + // A CronJob pod with no identity labels: the CronJob name, which comes before the Job name. + {"cronjob", false, map[string]string{ + "service.name": "otlp-e2e-cron", + "k8s.cronjob.name": "otlp-e2e-cron", + }}, +} + +// metricNames is every metric the OTLP pipelines export for this cluster: the sender's counter and the +// span metrics computed from its spans. +var metricNames = []string{ + otlpendpoint.MetricName, + "traces.span.metrics.calls", + "traces.span.metrics.duration", +} + +// TestEndpoint checks that the operator opened the default ports on the node agent's Service and host +// network, and that every post from every app sender pod, through the Service, was accepted. +func TestEndpoint(t *testing.T) { + ctx := context.Background() + servicePorts, err := otlpendpoint.ServicePorts(ctx, kube) + require.NoError(t, err) + assert.Contains(t, servicePorts, int32(otlpendpoint.GRPCPort), "Service port") + assert.Contains(t, servicePorts, int32(otlpendpoint.HTTPPort), "Service port") + + hostPorts, err := otlpendpoint.AgentHostPorts(ctx, kube) + require.NoError(t, err) + assert.Contains(t, hostPorts, int32(otlpendpoint.GRPCPort), "agent host port") + assert.Contains(t, hostPorts, int32(otlpendpoint.HTTPPort), "agent host port") + + // A post can fail while an agent pod restarts, so the check is that posts succeed now: the last one, + // and at least minAccepted in all. + const minAccepted = 3 + for _, pod := range appPods(t) { + statuses, err := otlpendpoint.SendStatuses(ctx, kube, senderNamespace, pod.Name) + require.NoError(t, err) + for _, signal := range otlpendpoint.Signals { + posts := statuses[signal] + require.NotEmpty(t, posts, "%s: no %s posts in the sender log", pod.Name, signal) + accepted := 0 + for _, status := range posts { + if status == "200" { + accepted++ + } + } + assert.GreaterOrEqual(t, accepted, minAccepted, "%s: accepted %s posts of %v", pod.Name, signal, posts) + assert.Equal(t, "200", posts[len(posts)-1], "%s: last %s post", pod.Name, signal) + } + } +} + +// TestMetricResourceAttributes checks the enrichment of the app sender's metric from each of its pods, +// one per node. The values the agent can only get right by resolving the right pod (its name, UID, +// node and that node's instance) prove that the source IP survived the Service. +func TestMetricResourceAttributes(t *testing.T) { + for _, pod := range appPods(t) { + what := fmt.Sprintf("%s from %s", otlpendpoint.MetricName, pod.Name) + results := eventually(t, what, func() ([]otelmetrics.MetricResult, error) { + return query(otlpendpoint.MetricName, map[string]string{"@resource.k8s.pod.name": pod.Name}) + }) + for _, r := range results { + assertResource(t, what, expectedResource(t, pod), toAny(r.Labels.Resource)) + assert.Equal(t, otlpvalidation.CloudWatchSource, r.Labels.Instrumentation["cloudwatch.source"], what) + assert.Equal(t, otlpendpoint.Solution, r.Labels.Instrumentation["cloudwatch.solution"], what) + assert.Equal(t, "app", r.Labels.Datapoint[otlpendpoint.CaseAttribute], what) + } + } +} + +// TestServiceIdentity checks service.name and its neighbours for each identity case. The case is a +// data point attribute, which no processor changes. +// +// The query keeps only telemetry the agent matched to a pod (it has k8s.namespace.name). The first +// records a new pod sends can arrive before the agent's pod cache has the pod's IP, and are then +// exported without pod attributes; the identity rules have nothing to work with for those. +func TestServiceIdentity(t *testing.T) { + for _, c := range identityCases { + c := c + t.Run(c.senderCase, func(t *testing.T) { + results := eventually(t, c.senderCase, func() ([]otelmetrics.MetricResult, error) { + results, err := query(otlpendpoint.MetricName, map[string]string{ + otlpendpoint.CaseAttribute: c.senderCase, + "@resource.k8s.namespace.name": senderNamespace, + }) + if err != nil || (c.everyNode && !fromEveryNode(results)) { + return nil, err + } + return results, nil + }) + want := map[string]string{ + "service.namespace": senderNamespace, + "k8s.namespace.name": senderNamespace, + } + for k, v := range c.want { + want[k] = v + } + for _, r := range results { + for key, value := range want { + assert.Equal(t, value, r.Labels.Resource[key], "%s: %s", c.senderCase, key) + } + } + }) + } +} + +// TestSpanMetrics checks the span metrics computed from the app sender's server spans: the span's HTTP +// dimensions, and the sender's resource attributes, pod labels included. +func TestSpanMetrics(t *testing.T) { + for _, metric := range metricNames[1:] { + results := eventually(t, metric, func() ([]otelmetrics.MetricResult, error) { + return query(metric, map[string]string{ + "@resource.service.name": appService, + "http.request.method": "GET", + "http.route": "/e2e", + }) + }) + for _, r := range results { + assert.Equal(t, appDeployment, r.Labels.Resource["k8s.pod.label.app"], metric) + assert.Equal(t, senderNamespace, r.Labels.Resource["k8s.namespace.name"], metric) + assert.Equal(t, otlpendpoint.Solution, r.Labels.Instrumentation["cloudwatch.solution"], metric) + } + } +} + +// TestMetricNames checks that the OTLP pipelines export exactly the expected metrics for this cluster, +// so a metric that appears or stops is caught. +func TestMetricNames(t *testing.T) { + matcher := fmt.Sprintf(`{__name__=~".+","@resource.k8s.cluster.name"=%q,"@instrumentation.cloudwatch.solution"=%q}`, + cfg.ClusterName, otlpendpoint.Solution) + var got []string + for i := 0; i < attempts; i++ { + if i > 0 { + time.Sleep(retryInterval) + } + names, err := client.MetricNames(context.Background(), matcher, time.Now().Add(-time.Hour), time.Now()) + require.NoError(t, err) + got = names + if len(got) >= len(metricNames) { + break + } + } + assert.ElementsMatch(t, metricNames, got) +} + +// TestContainerInsightsUnchanged checks that OTel Container Insights still reports the sender pods, +// from the node agents and the cluster scraper, with its own scope, and that the attributes only the +// OTLP pipelines add did not reach it through the shared processors. +func TestContainerInsightsUnchanged(t *testing.T) { + inNamespace := map[string]string{"@resource.k8s.namespace.name": senderNamespace} + for _, metric := range []string{"container_cpu_usage_seconds_total", "k8s.pod.cpu.usage", "kube_pod_info"} { + results := eventually(t, metric, func() ([]otelmetrics.MetricResult, error) { + return query(metric, inNamespace) + }) + for _, r := range results { + assert.Equal(t, containerInsightsSolution, r.Labels.Instrumentation["cloudwatch.solution"], metric) + for _, key := range []string{"service.instance.id", "deployment.environment.name", "service.version"} { + assert.NotContains(t, r.Labels.Resource, key, metric) + } + } + } + + // The senders' stdout reaches the Container Insights application log group, and the OTLP log + // records do not. The absence is checked only once the group is known to receive the senders' logs. + group := fmt.Sprintf("/aws/otel/containerinsights/%s/application", cfg.ClusterName) + eventually(t, "sender stdout in "+group, func() ([]string, error) { + return insightsMessages(group, "otlp-e2e-send") + }) + otlpRecords, err := insightsMessages(group, otlpendpoint.LogBody("app")) + require.NoError(t, err) + assert.Empty(t, otlpRecords, "OTLP log records in %s", group) +} + +// TestLogs checks the app sender's OTLP log records: the log group and stream the chart derives, the +// record, its scope and the sender's resource attributes, from both pods. It also checks the stream of +// the unknown case, which is named after the replacement service.name. +func TestLogs(t *testing.T) { + group := otlpendpoint.LogGroup(cfg.ClusterName) + pods := map[string]corev1.Pod{} + for _, pod := range appPods(t) { + pods[pod.Name] = pod + } + + stream := strings.Join([]string{senderNamespace, senderNamespace, appService}, "/") + records := eventually(t, group+" "+stream, func() ([]otlpvalidation.LogRecord, error) { + records, err := streamRecords(group, stream) + if err != nil || !hasPods(records, pods) { + return nil, err + } + return records, nil + }) + for _, rec := range records { + podName := fmt.Sprint(rec.Resource.Attributes["k8s.pod.name"]) + pod, ok := pods[podName] + if !ok { + // A pod from before a restart: its attributes cannot be checked against a live pod. + continue + } + what := "log record from " + podName + assert.Equal(t, otlpendpoint.LogBody("app"), rec.Body, what) + assert.Equal(t, "INFO", rec.SeverityText, what) + assert.Equal(t, "app", fmt.Sprint(rec.Attributes[otlpendpoint.CaseAttribute]), what) + assert.Equal(t, otlpendpoint.Solution, fmt.Sprint(rec.Scope.Attributes["cloudwatch.solution"]), what) + assertResource(t, what, expectedResource(t, pod), rec.Resource.Attributes) + } + + unknown := strings.Join([]string{senderNamespace, senderNamespace, "otlp-e2e-unknown"}, "/") + eventually(t, group+" "+unknown, func() ([]otlpvalidation.LogRecord, error) { + return streamRecords(group, unknown) + }) + + if !t.Failed() { + awsservice.DeleteLogGroup(group) + } +} + +// TestTraces checks the app sender's spans in Transaction Search: the span, its scope and the sender's +// resource attributes, from both pods. +func TestTraces(t *testing.T) { + pods := map[string]corev1.Pod{} + for _, pod := range appPods(t) { + pods[pod.Name] = pod + } + q := fmt.Sprintf(`fields @message | filter @message like %q and @message like %q | limit 200`, cfg.ClusterName, appService) + spans := eventually(t, "spans of "+appService, func() ([]otlpvalidation.SpanRecord, error) { + messages, err := insightsMessages("aws/spans", "", q) + if err != nil { + return nil, err + } + var spans []otlpvalidation.SpanRecord + seen := map[string]bool{} + for _, m := range messages { + var s otlpvalidation.SpanRecord + if json.Unmarshal([]byte(m), &s) != nil { + continue + } + spans = append(spans, s) + seen[fmt.Sprint(s.Resource.Attributes["k8s.pod.name"])] = true + } + for name := range pods { + if !seen[name] { + return nil, nil + } + } + return spans, nil + }) + for _, s := range spans { + pod, ok := pods[fmt.Sprint(s.Resource.Attributes["k8s.pod.name"])] + if !ok { + continue + } + attrs := map[string]string{otlpendpoint.CaseAttribute: "app", "http.request.method": "GET"} + assert.NoError(t, otlpvalidation.AssertSpanContent(s, "GET /e2e", "SERVER", attrs, expectedResource(t, pod))) + assertAbsent(t, "span "+s.TraceID, s.Resource.Attributes) + } +} + +// expectedResource is the resource of the app sender's telemetry from pod: the Container Insights +// enrichment of the pod, its labels and its node's labels, and the service identity. +// otlpvalidation.PresenceOnly marks the values that cannot be known from the cluster. +func expectedResource(t *testing.T, pod corev1.Pod) map[string]string { + t.Helper() + node, ok := nodes[pod.Spec.NodeName] + require.True(t, ok, "pod %s is on unknown node %q", pod.Name, pod.Spec.NodeName) + replicaSet := "" + for _, owner := range pod.OwnerReferences { + if owner.Kind == "ReplicaSet" { + replicaSet = owner.Name + } + } + require.NotEmpty(t, replicaSet, "pod %s has no ReplicaSet owner", pod.Name) + return map[string]string{ + "k8s.cluster.name": cfg.ClusterName, + "k8s.namespace.name": senderNamespace, + "k8s.pod.name": pod.Name, + "k8s.pod.uid": string(pod.UID), + "k8s.pod.start_time": otlpvalidation.PresenceOnly, + "k8s.container.name": senderContainer, + "k8s.node.name": pod.Spec.NodeName, + "k8s.deployment.name": appDeployment, + "k8s.replicaset.name": replicaSet, + "k8s.workload.name": appDeployment, + "k8s.workload.type": "Deployment", + "cloud.provider": "aws", + "cloud.platform": "aws_eks", + "cloud.region": cfg.Region, + "cloud.account.id": cfg.AccountID, + "cloud.availability_zone": node.Labels["topology.kubernetes.io/zone"], + "cloud.resource_id": fmt.Sprintf("arn:aws:eks:%s:%s:cluster/%s", cfg.Region, cfg.AccountID, cfg.ClusterName), + "host.id": otlpendpoint.NodeInstanceID(node), + "host.type": node.Labels["node.kubernetes.io/instance-type"], + "host.name": otlpvalidation.PresenceOnly, + "host.image.id": otlpvalidation.PresenceOnly, + "service.name": appService, + "service.namespace": senderNamespace, + "service.version": appVersion, + "service.instance.id": strings.Join([]string{senderNamespace, pod.Name, senderContainer}, "."), + "deployment.environment.name": fmt.Sprintf("aws_eks:%s/%s", cfg.ClusterName, senderNamespace), + "k8s.pod.label.app": appDeployment, + "k8s.pod.label.app.kubernetes.io/name": appService, + "k8s.pod.label.app.kubernetes.io/version": appVersion, + "k8s.node.label.otlp-e2e-node": "v", + "k8s.node.label.kubernetes.io/os": "linux", + } +} + +// absentKeys are resource attributes that must not reach CloudWatch: labels on the chart's built-in +// exclusion list, which otelFilters applies by default. +var absentKeys = []string{ + "k8s.pod.label.pod-template-hash", + "k8s.node.label.kubernetes.io/hostname", + "k8s.node.label.topology.kubernetes.io/zone", + "k8s.node.label.node.kubernetes.io/instance-type", +} + +// absentPrefixes are resource attribute prefixes the pipelines use internally and remove before export. +var absentPrefixes = []string{"_tmp.", "aws.log."} + +func assertResource(t *testing.T, what string, want map[string]string, got map[string]any) { + t.Helper() + assert.NoError(t, otlpvalidation.AssertAttributes(what, want, got)) + assertAbsent(t, what, got) +} + +func assertAbsent(t *testing.T, what string, got map[string]any) { + t.Helper() + for _, key := range absentKeys { + assert.NotContains(t, got, key, what) + } + for key := range got { + for _, prefix := range absentPrefixes { + assert.False(t, strings.HasPrefix(key, prefix), "%s: internal attribute %s", what, key) + } + } +} + +// appPods returns the running pods of the app sender, which must be on every node. +func appPods(t *testing.T) []corev1.Pod { + t.Helper() + pods, err := otlpendpoint.RunningPods(context.Background(), kube, senderNamespace, "app="+appDeployment) + require.NoError(t, err) + onNodes := map[string]bool{} + for _, pod := range pods { + onNodes[pod.Spec.NodeName] = true + } + require.Len(t, onNodes, len(nodes), "app sender pods must run on every node") + return pods +} + +func query(metric string, labels map[string]string) ([]otelmetrics.MetricResult, error) { + selector := fmt.Sprintf(`{"__name__"=%q,"@resource.k8s.cluster.name"=%q`, metric, cfg.ClusterName) + keys := make([]string, 0, len(labels)) + for key := range labels { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + selector += fmt.Sprintf(`,%q=%q`, key, labels[key]) + } + return client.Query(context.Background(), selector+"}") +} + +func fromEveryNode(results []otelmetrics.MetricResult) bool { + reported := map[string]bool{} + for _, r := range results { + reported[r.Labels.Resource["k8s.node.name"]] = true + } + for name := range nodes { + if !reported[name] { + return false + } + } + return true +} + +// streamRecords returns the OTLP log records of a log stream from the last lookback. +func streamRecords(group, stream string) ([]otlpvalidation.LogRecord, error) { + since, until := time.Now().Add(-lookback), time.Now() + events, err := awsservice.GetLogsSince(group, stream, &since, &until) + if err != nil { + return nil, err + } + records := make([]otlpvalidation.LogRecord, 0, len(events)) + for _, event := range events { + var rec otlpvalidation.LogRecord + if err := json.Unmarshal([]byte(aws.ToString(event.Message)), &rec); err != nil { + return nil, fmt.Errorf("unparseable OTLP log record in %s/%s: %w", group, stream, err) + } + records = append(records, rec) + } + return records, nil +} + +func hasPods(records []otlpvalidation.LogRecord, pods map[string]corev1.Pod) bool { + seen := map[string]bool{} + for _, rec := range records { + seen[fmt.Sprint(rec.Resource.Attributes["k8s.pod.name"])] = true + } + for name := range pods { + if !seen[name] { + return false + } + } + return true +} + +// insightsMessages runs a Logs Insights query over the last lookback and returns the @message of each +// row. With an empty query, it returns the messages that contain marker. +func insightsMessages(group, marker string, query ...string) ([]string, error) { + q := fmt.Sprintf("fields @message | filter @message like %q | limit 1000", marker) + if len(query) > 0 { + q = query[0] + } + end := time.Now() + rows, err := awsservice.GetLogQueryResults(group, end.Add(-lookback).Unix(), end.Unix(), q) + if err != nil { + return nil, err + } + var messages []string + for _, row := range rows { + for _, field := range row { + if aws.ToString(field.Field) == "@message" { + messages = append(messages, aws.ToString(field.Value)) + } + } + } + return messages, nil +} + +func toAny(m map[string]string) map[string]any { + out := make(map[string]any, len(m)) + for k, v := range m { + out[k] = v + } + return out +} + +// eventually retries fetch until it returns a non-empty result and returns it. If every attempt fails +// or is empty, it stops the test, so the checks that follow run only on data that arrived. +func eventually[T any](t *testing.T, what string, fetch func() ([]T, error)) []T { + t.Helper() + var result []T + var err error + for i := 0; i < attempts; i++ { + if i > 0 { + time.Sleep(retryInterval) + } + result, err = fetch() + if err == nil && len(result) > 0 { + return result + } + } + require.NoError(t, err, "querying %s", what) + require.NotEmpty(t, result, "no %s after %d attempts", what, attempts) + return result +} diff --git a/test/otel/otlp/setup_test.go b/test/otel/otlp/setup_test.go new file mode 100644 index 000000000..995c770e2 --- /dev/null +++ b/test/otel/otlp/setup_test.go @@ -0,0 +1,111 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otlp + +import ( + "context" + "flag" + "fmt" + "os" + "testing" + "time" + + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sts" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + + "github.com/aws/amazon-cloudwatch-agent-test/environment" + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" + "github.com/aws/amazon-cloudwatch-agent-test/util/awsservice" + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +var ( + cfg otelmetrics.TestConfig + client *otelmetrics.OtelMetricsClient + kube kubernetes.Interface + // nodes are the cluster's nodes, keyed by name. Resource attributes that come from the node (host, + // zone, labels) are checked against them. + nodes map[string]corev1.Node +) + +func TestMain(m *testing.M) { + environment.RegisterEnvironmentMetaDataFlags() + flag.Parse() + env := environment.GetEnvironmentMetaData() + + region := env.Region + if region == "" { + region = os.Getenv("AWS_REGION") + } + if region == "" { + fmt.Fprintf(os.Stderr, "Region not set\n") + os.Exit(1) + } + + clusterName := env.EKSClusterName + if clusterName == "" { + clusterName = os.Getenv("CLUSTER_NAME") + } + if clusterName == "" { + fmt.Fprintf(os.Stderr, "Cluster name not set\n") + os.Exit(1) + } + + ctx := context.Background() + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) + if err != nil { + fmt.Fprintf(os.Stderr, "AWS config error: %v\n", err) + os.Exit(1) + } + identity, err := sts.NewFromConfig(awsCfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + if err != nil { + fmt.Fprintf(os.Stderr, "STS GetCallerIdentity error: %v\n", err) + os.Exit(1) + } + + cfg = otelmetrics.TestConfig{ + Region: region, + Endpoint: fmt.Sprintf("https://monitoring.%s.amazonaws.com", region), + Timeout: 30 * time.Second, + MaxRetries: 3, + ClusterName: clusterName, + AccountID: *identity.Account, + SigningService: "monitoring", + } + if client, err = otelmetrics.NewClient(ctx, cfg); err != nil { + fmt.Fprintf(os.Stderr, "Client error: %v\n", err) + os.Exit(1) + } + if err := awsservice.ConfigureAWSClients(region); err != nil { + fmt.Fprintf(os.Stderr, "AWS clients error: %v\n", err) + os.Exit(1) + } + + clientset, err := otlpendpoint.KubeClient() + if err != nil { + fmt.Fprintf(os.Stderr, "Kubernetes client error: %v\n", err) + os.Exit(1) + } + kube = clientset + list, err := kube.CoreV1().Nodes().List(ctx, metav1.ListOptions{}) + if err != nil { + fmt.Fprintf(os.Stderr, "Kubernetes nodes error: %v\n", err) + os.Exit(1) + } + if len(list.Items) == 0 { + fmt.Fprintf(os.Stderr, "Kubernetes nodes error: no nodes\n") + os.Exit(1) + } + nodes = make(map[string]corev1.Node, len(list.Items)) + for _, node := range list.Items { + nodes[node.Name] = node + } + + os.Exit(m.Run()) +} diff --git a/test/otel/otlp_disabled/otlp_disabled_test.go b/test/otel/otlp_disabled/otlp_disabled_test.go new file mode 100644 index 000000000..996e9a2b6 --- /dev/null +++ b/test/otel/otlp_disabled/otlp_disabled_test.go @@ -0,0 +1,87 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +// Package otlp_disabled tests the Helm chart with otlp.enabled=false next to OTel Container Insights +// (terraform/eks/daemon/otel-otlp-disabled): the node agent serves no OTLP endpoint, so a pod that +// sends OTLP through the cloudwatch-agent Service fails and nothing reaches CloudWatch. +package otlp_disabled + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" +) + +const ( + probeNamespace = "otlp-e2e" + probeSelector = "app=otlp-e2e-probe" + + // minPosts is the number of posts per signal the probe must have made before its failures count. + minPosts = 3 + + attempts = 10 + retryInterval = 30 * time.Second +) + +// TestNoOTLPPorts checks that the operator opened neither default OTLP port on the node agent's +// Service or host network. +func TestNoOTLPPorts(t *testing.T) { + ctx := context.Background() + servicePorts, err := otlpendpoint.ServicePorts(ctx, kube) + require.NoError(t, err) + hostPorts, err := otlpendpoint.AgentHostPorts(ctx, kube) + require.NoError(t, err) + for _, port := range []int32{otlpendpoint.GRPCPort, otlpendpoint.HTTPPort} { + assert.NotContains(t, servicePorts, port, "Service port") + assert.NotContains(t, hostPorts, port, "agent host port") + } +} + +// TestSendFails checks that no post from the probe pod through the Service succeeded. +func TestSendFails(t *testing.T) { + ctx := context.Background() + pods, err := otlpendpoint.RunningPods(ctx, kube, probeNamespace, probeSelector) + require.NoError(t, err) + require.NotEmpty(t, pods, "no running probe pod") + for _, pod := range pods { + statuses, err := otlpendpoint.SendStatuses(ctx, kube, probeNamespace, pod.Name) + require.NoError(t, err) + for _, signal := range otlpendpoint.Signals { + require.GreaterOrEqual(t, len(statuses[signal]), minPosts, "%s: %s posts in the probe log", pod.Name, signal) + for _, status := range statuses[signal] { + assert.Equal(t, "000", status, "%s: %s post reached a server", pod.Name, signal) + } + } + } +} + +// TestNoOTLPTelemetry checks that no metric with the OTLP pipelines' scope arrived for this cluster. +// It runs only once OTel Container Insights metrics from the node agents are seen, so that it cannot +// pass because the agents export nothing. +func TestNoOTLPTelemetry(t *testing.T) { + ctx := context.Background() + ci := fmt.Sprintf(`{"__name__"="container_cpu_usage_seconds_total","@resource.k8s.cluster.name"=%q}`, cfg.ClusterName) + for i := 0; ; i++ { + results, err := client.Query(ctx, ci) + require.NoError(t, err) + if len(results) > 0 { + break + } + require.Less(t, i, attempts, "no OTel Container Insights metrics for the cluster") + time.Sleep(retryInterval) + } + + matcher := fmt.Sprintf(`{__name__=~".+","@resource.k8s.cluster.name"=%q,"@instrumentation.cloudwatch.solution"=%q}`, + cfg.ClusterName, otlpendpoint.Solution) + names, err := client.MetricNames(ctx, matcher, time.Now().Add(-time.Hour), time.Now()) + require.NoError(t, err) + assert.Empty(t, names, "metrics with the OTLP scope") +} diff --git a/test/otel/otlp_disabled/setup_test.go b/test/otel/otlp_disabled/setup_test.go new file mode 100644 index 000000000..7f4f2429f --- /dev/null +++ b/test/otel/otlp_disabled/setup_test.go @@ -0,0 +1,92 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otlp_disabled + +import ( + "context" + "flag" + "fmt" + "os" + "testing" + "time" + + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/service/sts" + "k8s.io/client-go/kubernetes" + + "github.com/aws/amazon-cloudwatch-agent-test/environment" + "github.com/aws/amazon-cloudwatch-agent-test/test/otel/otlpendpoint" + "github.com/aws/amazon-cloudwatch-agent-test/util/awsservice" + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +var ( + cfg otelmetrics.TestConfig + client *otelmetrics.OtelMetricsClient + kube kubernetes.Interface +) + +func TestMain(m *testing.M) { + environment.RegisterEnvironmentMetaDataFlags() + flag.Parse() + env := environment.GetEnvironmentMetaData() + + region := env.Region + if region == "" { + region = os.Getenv("AWS_REGION") + } + if region == "" { + fmt.Fprintf(os.Stderr, "Region not set\n") + os.Exit(1) + } + + clusterName := env.EKSClusterName + if clusterName == "" { + clusterName = os.Getenv("CLUSTER_NAME") + } + if clusterName == "" { + fmt.Fprintf(os.Stderr, "Cluster name not set\n") + os.Exit(1) + } + + ctx := context.Background() + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, awsconfig.WithRegion(region)) + if err != nil { + fmt.Fprintf(os.Stderr, "AWS config error: %v\n", err) + os.Exit(1) + } + identity, err := sts.NewFromConfig(awsCfg).GetCallerIdentity(ctx, &sts.GetCallerIdentityInput{}) + if err != nil { + fmt.Fprintf(os.Stderr, "STS GetCallerIdentity error: %v\n", err) + os.Exit(1) + } + + cfg = otelmetrics.TestConfig{ + Region: region, + Endpoint: fmt.Sprintf("https://monitoring.%s.amazonaws.com", region), + Timeout: 30 * time.Second, + MaxRetries: 3, + ClusterName: clusterName, + AccountID: *identity.Account, + SigningService: "monitoring", + } + if client, err = otelmetrics.NewClient(ctx, cfg); err != nil { + fmt.Fprintf(os.Stderr, "Client error: %v\n", err) + os.Exit(1) + } + if err := awsservice.ConfigureAWSClients(region); err != nil { + fmt.Fprintf(os.Stderr, "AWS clients error: %v\n", err) + os.Exit(1) + } + + clientset, err := otlpendpoint.KubeClient() + if err != nil { + fmt.Fprintf(os.Stderr, "Kubernetes client error: %v\n", err) + os.Exit(1) + } + kube = clientset + os.Exit(m.Run()) +} diff --git a/test/otel/otlpendpoint/otlpendpoint.go b/test/otel/otlpendpoint/otlpendpoint.go new file mode 100644 index 000000000..0e051d35d --- /dev/null +++ b/test/otel/otlpendpoint/otlpendpoint.go @@ -0,0 +1,158 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +// Package otlpendpoint holds the helpers shared by the suites that test the Helm chart's OTLP endpoint +// on EKS: test/otel/otlp, test/otel/otlp_disabled and the OTLP cases of test/otel/filters. The senders +// they read run terraform/otlp_service_sender.sh. +package otlpendpoint + +import ( + "bufio" + "context" + "fmt" + "os" + "path/filepath" + "strings" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/clientcmd" +) + +const ( + // AgentNamespace and AgentName are the chart's namespace and node agent. The node agent's Service, + // which the operator creates from the agent's receivers, has the agent's name. + AgentNamespace = "amazon-cloudwatch" + AgentName = "cloudwatch-agent" + + // GRPCPort and HTTPPort are the chart's default OTLP ports. + GRPCPort = 4317 + HTTPPort = 4318 + + // Solution is the cloudwatch.solution scope attribute the chart's OTLP pipelines set. + Solution = "otel-otlp" + + // MetricName is the counter otlp_service_sender.sh sends by default, and CaseAttribute the attribute + // it sets on every data point, log record and span. + MetricName = "otlp_e2e_requests" + CaseAttribute = "otlp_e2e_case" + + sendPrefix = "otlp-e2e-send " +) + +// Signals are the OTLP signals otlp_service_sender.sh posts, in the names of their /v1 paths. +var Signals = []string{"metrics", "logs", "traces"} + +// LogBody is the body of every log record otlp_service_sender.sh sends for case. +func LogBody(senderCase string) string { + return "otlp-e2e " + senderCase +} + +// LogGroup is the log group the chart's OTLP logs pipeline writes to when the sender sets none. +func LogGroup(cluster string) string { + return fmt.Sprintf("/aws/cwagent/%s/otlp", cluster) +} + +// KubeClient returns a client for the cluster in the kubeconfig that Terraform writes for kubectl. +func KubeClient() (*kubernetes.Clientset, error) { + kubeconfig := os.Getenv("KUBECONFIG") + if kubeconfig == "" { + kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config") + } + restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfig) + if err != nil { + return nil, err + } + return kubernetes.NewForConfig(restConfig) +} + +// ServicePorts returns the port numbers of the node agent's Service, or none if there is no Service. +func ServicePorts(ctx context.Context, cs kubernetes.Interface) ([]int32, error) { + svc, err := cs.CoreV1().Services(AgentNamespace).Get(ctx, AgentName, metav1.GetOptions{}) + if apierrors.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, err + } + ports := make([]int32, 0, len(svc.Spec.Ports)) + for _, p := range svc.Spec.Ports { + ports = append(ports, p.Port) + } + return ports, nil +} + +// AgentHostPorts returns the ports the node agent DaemonSet's containers open on the node. The agent +// runs on the host network, where a container port is a host port; otherwise only declared host ports +// count. +func AgentHostPorts(ctx context.Context, cs kubernetes.Interface) ([]int32, error) { + ds, err := cs.AppsV1().DaemonSets(AgentNamespace).Get(ctx, AgentName, metav1.GetOptions{}) + if err != nil { + return nil, err + } + return hostPorts(ds.Spec.Template.Spec), nil +} + +func hostPorts(spec corev1.PodSpec) []int32 { + var ports []int32 + for _, c := range spec.Containers { + for _, p := range c.Ports { + switch { + case spec.HostNetwork: + ports = append(ports, p.ContainerPort) + case p.HostPort != 0: + ports = append(ports, p.HostPort) + } + } + } + return ports +} + +// RunningPods returns the running pods in namespace that match the label selector. +func RunningPods(ctx context.Context, cs kubernetes.Interface, namespace, selector string) ([]corev1.Pod, error) { + list, err := cs.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector}) + if err != nil { + return nil, err + } + var pods []corev1.Pod + for _, pod := range list.Items { + if pod.Status.Phase == corev1.PodRunning { + pods = append(pods, pod) + } + } + return pods, nil +} + +// SendStatuses reads the log of a sender pod and returns, per signal, the HTTP status of every post +// ("000" when the connection failed). +func SendStatuses(ctx context.Context, cs kubernetes.Interface, namespace, pod string) (map[string][]string, error) { + raw, err := cs.CoreV1().Pods(namespace).GetLogs(pod, &corev1.PodLogOptions{}).DoRaw(ctx) + if err != nil { + return nil, err + } + return ParseSendStatuses(string(raw)), nil +} + +// ParseSendStatuses parses the "otlp-e2e-send " lines of a sender log. +func ParseSendStatuses(log string) map[string][]string { + statuses := map[string][]string{} + scanner := bufio.NewScanner(strings.NewReader(log)) + for scanner.Scan() { + line, ok := strings.CutPrefix(scanner.Text(), sendPrefix) + if !ok { + continue + } + if fields := strings.Fields(line); len(fields) == 2 { + statuses[fields[0]] = append(statuses[fields[0]], fields[1]) + } + } + return statuses +} + +// NodeInstanceID returns the EC2 instance ID from a node's provider ID (aws:////). +func NodeInstanceID(node corev1.Node) string { + parts := strings.Split(node.Spec.ProviderID, "/") + return parts[len(parts)-1] +} diff --git a/test/otel/otlpendpoint/otlpendpoint_test.go b/test/otel/otlpendpoint/otlpendpoint_test.go new file mode 100644 index 000000000..9e2c8a412 --- /dev/null +++ b/test/otel/otlpendpoint/otlpendpoint_test.go @@ -0,0 +1,40 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otlpendpoint + +import ( + "testing" + + "github.com/stretchr/testify/assert" + corev1 "k8s.io/api/core/v1" +) + +func TestParseSendStatuses(t *testing.T) { + log := "otlp-e2e-send metrics 200\n" + + "otlp-e2e-send logs 200\n" + + "some other line\n" + + "otlp-e2e-send traces 000\n" + + "otlp-e2e-send metrics 503\n" + + "otlp-e2e-send malformed\n" + assert.Equal(t, map[string][]string{ + "metrics": {"200", "503"}, + "logs": {"200"}, + "traces": {"000"}, + }, ParseSendStatuses(log)) + assert.Empty(t, ParseSendStatuses("")) +} + +func TestNodeInstanceID(t *testing.T) { + node := corev1.Node{Spec: corev1.NodeSpec{ProviderID: "aws:///us-west-2a/i-0123456789abcdef0"}} + assert.Equal(t, "i-0123456789abcdef0", NodeInstanceID(node)) +} + +func TestHostPorts(t *testing.T) { + containers := []corev1.Container{{Ports: []corev1.ContainerPort{ + {ContainerPort: 4317}, + {ContainerPort: 8080, HostPort: 18080}, + }}} + assert.Equal(t, []int32{4317, 8080}, hostPorts(corev1.PodSpec{HostNetwork: true, Containers: containers})) + assert.Equal(t, []int32{18080}, hostPorts(corev1.PodSpec{Containers: containers})) +}