Support JMX annotations.

Author: jefchien

internal/manifests/collector/adapters/config_from.go

@@ -58,6 +58,7 @@ type Traces struct {
 type MetricsCollected struct {
 	StatsD   *statsD   `json:"statsd,omitempty"`
 	CollectD *collectD `json:"collectd,omitempty"`
+	JMX      *jmx      `json:"jmx,omitempty"`
 }
 
 type LogMetricsCollected struct {
@@ -86,6 +87,9 @@ type AppSignals struct {
 
 type emf struct {
 }
+
+type jmx struct{}
+
 type kubernetes struct {
 	EnhancedContainerInsights bool `json:"enhanced_container_insights,omitempty"`
 	AcceleratedComputeMetrics bool `json:"accelerated_compute_metrics,omitempty"`

internal/manifests/collector/ports.go

@@ -27,22 +27,24 @@ const (
 	XrayTraces        = "aws-traces"
 	OtlpGrpc          = "otlp-grpc"
 	OtlpHttp          = "otlp-http"
-	AppSignalsGrpc    = "appsig-grpc"
-	AppSignalsHttp    = "appsig-http"
-	AppSignalsProxy   = "appsig-xray"
+	AppSignalsGrpc    = "appsignals-grpc"
+	AppSignalsHttp    = "appsignals-http"
+	AppSignalsProxy   = "appsignals-xray"
 	AppSignalsGrpcSA  = ":4315"
 	AppSignalsHttpSA  = ":4316"
 	AppSignalsProxySA = ":2000"
 	EMF               = "emf"
 	EMFTcp            = "emf-tcp"
 	EMFUdp            = "emf-udp"
 	CWA               = "cwa-"
+	JmxHttp           = "jmx-http"
 )
 
 var receiverDefaultPortsMap = map[string]int32{
 	StatsD:     8125,
 	CollectD:   25826,
 	XrayTraces: 2000,
+	JmxHttp:    4314,
 	OtlpGrpc:   4317,
 	OtlpHttp:   4318,
 	EMF:        25888,
@@ -83,6 +85,7 @@ func getContainerPorts(logger logr.Logger, cfg string, specPorts []corev1.Servic
 	if err != nil {
 		logger.Error(err, "error parsing cw agent config")
 	} else {
+		logger.Info("%v", config)
 		servicePorts = getServicePortsFromCWAgentConfig(logger, config)
 	}
 
@@ -143,6 +146,9 @@ func getMetricsReceiversServicePorts(logger logr.Logger, config *adapters.CwaCon
 	if config.Metrics.MetricsCollected.CollectD != nil {
 		getReceiverServicePort(logger, config.Metrics.MetricsCollected.CollectD.ServiceAddress, CollectD, corev1.ProtocolUDP, servicePortsMap)
 	}
+	if config.Metrics.MetricsCollected.JMX != nil {
+		getReceiverServicePort(logger, "", JmxHttp, corev1.ProtocolTCP, servicePortsMap)
+	}
 }
 
 func getReceiverServicePort(logger logr.Logger, serviceAddress string, receiverName string, protocol corev1.Protocol, servicePortsMap map[int32][]corev1.ServicePort) {

internal/manifests/collector/ports_test.go

@@ -204,6 +204,21 @@ func TestInvalidConfigGetContainerPorts(t *testing.T) {
 
 }
 
+func TestJMXGetContainerPorts(t *testing.T) {
+	cfg := getJSONStringFromFile("testdata/jmx.json")
+	containerPorts := getContainerPorts(logger, cfg, []corev1.ServicePort{})
+	assert.Equal(t, 4, len(containerPorts))
+	assert.Equal(t, int32(4315), containerPorts[AppSignalsGrpc].ContainerPort)
+	assert.Equal(t, AppSignalsGrpc, containerPorts[AppSignalsGrpc].Name)
+	assert.Equal(t, int32(4316), containerPorts[AppSignalsHttp].ContainerPort)
+	assert.Equal(t, AppSignalsHttp, containerPorts[AppSignalsHttp].Name)
+	assert.Equal(t, int32(2000), containerPorts[AppSignalsProxy].ContainerPort)
+	assert.Equal(t, AppSignalsProxy, containerPorts[AppSignalsProxy].Name)
+	assert.Equal(t, int32(4314), containerPorts[JmxHttp].ContainerPort)
+	assert.Equal(t, JmxHttp, containerPorts[JmxHttp].Name)
+	assert.Equal(t, corev1.ProtocolTCP, containerPorts[JmxHttp].Protocol)
+}
+
 func getJSONStringFromFile(path string) string {
 	buf, err := os.ReadFile(path)
 	if err != nil {

internal/manifests/collector/testdata/jmx.json

@@ -0,0 +1,8 @@
+{
+  "metrics": {
+    "metrics_collected": {
+      "jmx": {
+      }
+    }
+  }
+}

pkg/instrumentation/defaultinstrumentation.go

@@ -8,13 +8,13 @@ import (
 	"fmt"
 	"os"
 
-	"k8s.io/apimachinery/pkg/api/resource"
-
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/aws/amazon-cloudwatch-agent-operator/apis/v1alpha1"
 	"github.com/aws/amazon-cloudwatch-agent-operator/internal/manifests/collector/adapters"
+	"github.com/aws/amazon-cloudwatch-agent-operator/pkg/instrumentation/jmx"
 )
 
 const (
@@ -50,7 +50,7 @@ func getInstrumentationConfigForResource(langStr string, resourceStr string) cor
 	return instrumentationConfigForResource
 }
 
-func getDefaultInstrumentation(agentConfig *adapters.CwaConfig, isWindowsPod bool) (*v1alpha1.Instrumentation, error) {
+func getDefaultInstrumentation(agentConfig *adapters.CwaConfig, additionalEnvs map[Type]map[string]string, isWindowsPod bool) (*v1alpha1.Instrumentation, error) {
 	javaInstrumentationImage, ok := os.LookupEnv("AUTO_INSTRUMENTATION_JAVA")
 	if !ok {
 		return nil, errors.New("unable to determine java instrumentation image")
@@ -103,18 +103,7 @@ func getDefaultInstrumentation(agentConfig *adapters.CwaConfig, isWindowsPod boo
 			},
 			Java: v1alpha1.Java{
 				Image: javaInstrumentationImage,
-				Env: []corev1.EnvVar{
-					{Name: "OTEL_AWS_APP_SIGNALS_ENABLED", Value: "true"}, //TODO: remove in favor of new name once safe
-					{Name: "OTEL_AWS_APPLICATION_SIGNALS_ENABLED", Value: "true"},
-					{Name: "OTEL_TRACES_SAMPLER_ARG", Value: fmt.Sprintf("endpoint=%s://%s:2000", http, cloudwatchAgentServiceEndpoint)},
-					{Name: "OTEL_TRACES_SAMPLER", Value: "xray"},
-					{Name: "OTEL_EXPORTER_OTLP_PROTOCOL", Value: "http/protobuf"},
-					{Name: "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/traces", exporterPrefix, cloudwatchAgentServiceEndpoint)},
-					{Name: "OTEL_AWS_APP_SIGNALS_EXPORTER_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/metrics", exporterPrefix, cloudwatchAgentServiceEndpoint)}, //TODO: remove in favor of new name once safe
-					{Name: "OTEL_AWS_APPLICATION_SIGNALS_EXPORTER_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/metrics", exporterPrefix, cloudwatchAgentServiceEndpoint)},
-					{Name: "OTEL_METRICS_EXPORTER", Value: "none"},
-					{Name: "OTEL_LOGS_EXPORTER", Value: "none"},
-				},
+				Env:   getJavaEnvs(cloudwatchAgentServiceEndpoint, exporterPrefix, additionalEnvs[TypeJava]),
 				Resources: corev1.ResourceRequirements{
 					Limits:   getInstrumentationConfigForResource(java, limit),
 					Requests: getInstrumentationConfigForResource(java, request),
@@ -182,3 +171,38 @@ func getDefaultInstrumentation(agentConfig *adapters.CwaConfig, isWindowsPod boo
 		},
 	}, nil
 }
+
+func getJavaEnvs(cloudwatchAgentServiceEndpoint, exporterPrefix string, additionalEnvs map[string]string) []corev1.EnvVar {
+	envs := []corev1.EnvVar{
+		{Name: "OTEL_AWS_APP_SIGNALS_ENABLED", Value: "true"}, //TODO: remove in favor of new name once safe
+		{Name: "OTEL_AWS_APPLICATION_SIGNALS_ENABLED", Value: "true"},
+		{Name: "OTEL_TRACES_SAMPLER_ARG", Value: fmt.Sprintf("endpoint=%s://%s:2000", http, cloudwatchAgentServiceEndpoint)},
+		{Name: "OTEL_TRACES_SAMPLER", Value: "xray"},
+		{Name: "OTEL_EXPORTER_OTLP_PROTOCOL", Value: "http/protobuf"},
+		{Name: "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/traces", exporterPrefix, cloudwatchAgentServiceEndpoint)},
+		{Name: "OTEL_AWS_APP_SIGNALS_EXPORTER_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/metrics", exporterPrefix, cloudwatchAgentServiceEndpoint)}, //TODO: remove in favor of new name once safe
+		{Name: "OTEL_AWS_APPLICATION_SIGNALS_EXPORTER_ENDPOINT", Value: fmt.Sprintf("%s://%s:4316/v1/metrics", exporterPrefix, cloudwatchAgentServiceEndpoint)},
+	}
+	var jmxEnvs []corev1.EnvVar
+	if targetSystems, ok := additionalEnvs[jmx.EnvTargetSystem]; ok {
+		jmxEnvs = []corev1.EnvVar{
+			{Name: "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", Value: fmt.Sprintf("%s://%s:4314/v1/metrics", http, cloudwatchAgentServiceEndpoint)},
+			{Name: "OTEL_INSTRUMENTATION_RUNTIME_TELEMETRY_ENABLED", Value: "false"},
+			{Name: "OTEL_INSTRUMENTATION_COMMON_DEFAULT_ENABLED", Value: "false"},
+			{Name: "OTEL_JMX_ENABLED", Value: "true"},
+			{Name: "OTEL_JMX_TARGET_SYSTEM", Value: targetSystems},
+			{Name: "OTEL_EXPERIMENTAL_METRICS_VIEW_CONFIG", Value: "classpath:/jmx/view.yaml"},
+		}
+	}
+	if len(jmxEnvs) == 0 {
+		envs = append(
+			envs,
+			corev1.EnvVar{Name: "OTEL_METRICS_EXPORTER", Value: "none"},
+			corev1.EnvVar{Name: "OTEL_LOGS_EXPORTER", Value: "none"},
+		)
+	} else {
+		envs = append(envs, corev1.EnvVar{Name: "OTEL_LOGS_EXPORTER", Value: "none"})
+		envs = append(envs, jmxEnvs...)
+	}
+	return envs
+}

pkg/instrumentation/defaultinstrumentation_test.go

@@ -327,7 +327,7 @@ func Test_getDefaultInstrumentationLinux(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := getDefaultInstrumentation(tt.args.agentConfig, false)
+			got, err := getDefaultInstrumentation(tt.args.agentConfig, nil, false)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("getDefaultInstrumentation() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -650,7 +650,7 @@ func Test_getDefaultInstrumentationWindows(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := getDefaultInstrumentation(tt.args.agentConfig, true)
+			got, err := getDefaultInstrumentation(tt.args.agentConfig, nil, true)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("getDefaultInstrumentation() error = %v, wantErr %v", err, tt.wantErr)
 				return

pkg/instrumentation/jmx/jmx.go

@@ -0,0 +1,24 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package jmx
+
+const (
+	annotationPrefix = "cloudwatch.aws.amazon.com/inject-jmx-"
+)
+
+const (
+	EnvTargetSystem = "OTEL_JMX_TARGET_SYSTEM"
+
+	TargetJVM           = "jvm"
+	TargetTomcat        = "tomcat"
+	TargetKafka         = "kafka"
+	TargetKafkaConsumer = "kafka-consumer"
+	TargetKafkaProducer = "kafka-producer"
+)
+
+var SupportedTargets = []string{TargetJVM, TargetTomcat, TargetKafka, TargetKafkaConsumer, TargetKafkaProducer}
+
+func AnnotationKey(target string) string {
+	return annotationPrefix + target
+}

pkg/instrumentation/podmutator.go

@@ -19,6 +19,7 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent-operator/internal/manifests/collector/adapters"
 	"github.com/aws/amazon-cloudwatch-agent-operator/internal/webhook/podmutation"
 	"github.com/aws/amazon-cloudwatch-agent-operator/pkg/featuregate"
+	"github.com/aws/amazon-cloudwatch-agent-operator/pkg/instrumentation/jmx"
 )
 
 const (
@@ -368,8 +369,19 @@ func (pm *instPodMutator) getInstrumentationInstance(ctx context.Context, ns cor
 		return nil, nil
 	}
 
+	var additionalEnvs map[Type]map[string]string
+	if instAnnotation == annotationInjectJava {
+		additionalEnvs = map[Type]map[string]string{}
+		targetSystems := getJmxTargetSystems(ns, pod)
+		if len(targetSystems) != 0 {
+			additionalEnvs[TypeJava] = map[string]string{
+				jmx.EnvTargetSystem: strings.Join(targetSystems, ","),
+			}
+		}
+	}
+
 	if strings.EqualFold(instValue, "true") {
-		return pm.selectInstrumentationInstanceFromNamespace(ctx, ns, isWindowsPod(pod))
+		return pm.selectInstrumentationInstanceFromNamespace(ctx, ns, additionalEnvs, isWindowsPod(pod))
 	}
 
 	var instNamespacedName types.NamespacedName
@@ -388,7 +400,7 @@ func (pm *instPodMutator) getInstrumentationInstance(ctx context.Context, ns cor
 	return otelInst, nil
 }
 
-func (pm *instPodMutator) selectInstrumentationInstanceFromNamespace(ctx context.Context, ns corev1.Namespace, isWindowsPod bool) (*v1alpha1.Instrumentation, error) {
+func (pm *instPodMutator) selectInstrumentationInstanceFromNamespace(ctx context.Context, ns corev1.Namespace, additionalEnvs map[Type]map[string]string, isWindowsPod bool) (*v1alpha1.Instrumentation, error) {
 	var otelInsts v1alpha1.InstrumentationList
 	if err := pm.Client.List(ctx, &otelInsts, client.InNamespace(ns.Name)); err != nil {
 		return nil, err
@@ -403,7 +415,7 @@ func (pm *instPodMutator) selectInstrumentationInstanceFromNamespace(ctx context
 			pm.Logger.Error(err, "unable to retrieve cloudwatch agent config for instrumentation")
 		}
 
-		return getDefaultInstrumentation(config, isWindowsPod)
+		return getDefaultInstrumentation(config, additionalEnvs, isWindowsPod)
 	case s > 1:
 		return nil, errMultipleInstancesPossible
 	default:
@@ -425,3 +437,14 @@ func GetAmazonCloudWatchAgentResource(ctx context.Context, c client.Client, name
 func isWindowsPod(pod corev1.Pod) bool {
 	return pod.Spec.NodeSelector["kubernetes.io/os"] == "windows"
 }
+
+func getJmxTargetSystems(ns corev1.Namespace, pod corev1.Pod) []string {
+	var targetSystems []string
+	for _, target := range jmx.SupportedTargets {
+		value := annotationValue(ns.ObjectMeta, pod.ObjectMeta, jmx.AnnotationKey(target))
+		if strings.EqualFold(value, "true") {
+			targetSystems = append(targetSystems, target)
+		}
+	}
+	return targetSystems
+}

pkg/instrumentation/podmutator_test.go

@@ -9,6 +9,8 @@ import (
 	"os"
 	"testing"
 
+	"github.com/aws/amazon-cloudwatch-agent-operator/pkg/instrumentation/jmx"
+
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -31,7 +33,7 @@ const (
 )
 
 func TestGetInstrumentationInstanceFromNameSpaceDefault(t *testing.T) {
-	defaultInst, _ := getDefaultInstrumentation(&adapters.CwaConfig{}, false)
+	defaultInst, _ := getDefaultInstrumentation(&adapters.CwaConfig{}, nil, false)
 	namespace := corev1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "default-namespace",
@@ -45,12 +47,60 @@ func TestGetInstrumentationInstanceFromNameSpaceDefault(t *testing.T) {
 		Client: fake.NewClientBuilder().Build(),
 		Logger: logr.Logger{},
 	}
-	instrumentation, err := podMutator.selectInstrumentationInstanceFromNamespace(context.Background(), namespace, false)
+	instrumentation, err := podMutator.selectInstrumentationInstanceFromNamespace(context.Background(), namespace, nil, false)
 
 	assert.Nil(t, err)
 	assert.Equal(t, defaultInst, instrumentation)
 }
 
+func TestGetInstrumentationInstanceJMX(t *testing.T) {
+	if err := v1alpha1.AddToScheme(testScheme); err != nil {
+		fmt.Printf("failed to register scheme: %v", err)
+		os.Exit(1)
+	}
+	mutator := instPodMutator{
+		Client: fake.NewClientBuilder().Build(),
+		Logger: logr.Discard(),
+	}
+
+	tests := []struct {
+		name    string
+		pod     corev1.Pod
+		ns      corev1.Namespace
+		wantLen int
+		wantEnv []corev1.EnvVar
+	}{
+		{
+			name: "enable jvm/tomcat",
+			pod: corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						annotationInjectJava:                "true",
+						jmx.AnnotationKey(jmx.TargetJVM):    "true",
+						jmx.AnnotationKey(jmx.TargetTomcat): "true",
+					},
+				},
+			},
+			ns:      corev1.Namespace{},
+			wantLen: 15,
+			wantEnv: []corev1.EnvVar{
+				{Name: "OTEL_JMX_ENABLED", Value: "true"},
+				{Name: "OTEL_JMX_TARGET_SYSTEM", Value: "jvm,tomcat"},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			inst, err := mutator.getInstrumentationInstance(context.Background(), tt.ns, tt.pod, annotationInjectJava)
+			assert.NoError(t, err)
+			assert.Len(t, inst.Spec.Java.Env, tt.wantLen)
+			for _, env := range tt.wantEnv {
+				assert.Containsf(t, inst.Spec.Java.Env, env, "java env does not contain %s:%s", env.Name, env.Value)
+			}
+		})
+	}
+}
+
 func TestMutatePod(t *testing.T) {
 	mutator := NewMutator(logr.Discard(), k8sClient, record.NewFakeRecorder(100))
 	require.NotNil(t, mutator)