diff --git a/generator/test_case_generator.go b/generator/test_case_generator.go index 4d446d50..57696981 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 4ea5932e..9ac816f4 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 6f103a66..d5945eb1 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 00000000..9c482543 --- /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 00000000..a562a7f1 --- /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 00000000..be5cbf6b --- /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 00000000..962d2ed8 --- /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 00000000..a562a7f1 --- /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 00000000..daa180f2 --- /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 00000000..f51c2c0c --- /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 00000000..6b81f055 --- /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 00000000..93743a8f --- /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 8ed55657..0b829309 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 00000000..07597ad7 --- /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 00000000..feff56d2 --- /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 00000000..995c770e --- /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 00000000..996e9a2b --- /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 00000000..7f4f2429 --- /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 00000000..0e051d35 --- /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 00000000..9e2c8a41 --- /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})) +}