|
| 1 | +// Copyright The OpenTelemetry Authors |
| 2 | +// SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +package receivercreator // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator" |
| 5 | + |
| 6 | +import ( |
| 7 | + "fmt" |
| 8 | + |
| 9 | + "go.uber.org/zap" |
| 10 | + |
| 11 | + "github.com/open-telemetry/opentelemetry-collector-contrib/extension/observer" |
| 12 | +) |
| 13 | + |
| 14 | +const ( |
| 15 | + otelHints = "io.opentelemetry.collector.receiver-creator" |
| 16 | + metricsHint = "metrics" |
| 17 | + hintsMetricsReceiver = "receiver" |
| 18 | + hintsMetricsEndpoint = "endpoint" |
| 19 | + hintsMetricsCollectionInterval = "collection_interval" |
| 20 | + hintsMetricsTimeout = "timeout" |
| 21 | + hintsMetricsUsername = "username" |
| 22 | + hintsMetricsPassword = "password" |
| 23 | +) |
| 24 | + |
| 25 | +// HintsTemplatesBuilder creates configuration templates from provided hints. |
| 26 | +type HintsTemplatesBuilder interface { |
| 27 | + createReceiverTemplatesFromHints() ([]receiverTemplate, error) |
| 28 | +} |
| 29 | + |
| 30 | +// K8sHintsBuilder creates configurations from hints provided as Pod's annotations. |
| 31 | +type K8sHintsBuilder struct { |
| 32 | + logger *zap.Logger |
| 33 | + config K8sHintsConfig |
| 34 | +} |
| 35 | + |
| 36 | +// createReceiverTemplateFromHints creates a receiver configuration based on the provided hints. |
| 37 | +// Hints are extracted from Pod's annotations. |
| 38 | +// Metrics configurations are only created for Port Endpoints. |
| 39 | +// TODO: Logs configurations are only created for Pod Container Endpoints. |
| 40 | +func (builder *K8sHintsBuilder) createReceiverTemplateFromHints(env observer.EndpointEnv) (*receiverTemplate, error) { |
| 41 | + var endpointType string |
| 42 | + var podUID string |
| 43 | + var annotations map[string]string |
| 44 | + |
| 45 | + builder.logger.Debug("handling hints for added endpoint", zap.Any("env", env)) |
| 46 | + |
| 47 | + if pod, ok := env["pod"]; ok { |
| 48 | + endpointPod, ok := pod.(observer.EndpointEnv) |
| 49 | + if !ok { |
| 50 | + return nil, fmt.Errorf("could not extract endpoint's pod: %v", zap.Any("endpointPod", pod)) |
| 51 | + } |
| 52 | + ann := endpointPod["annotations"] |
| 53 | + if ann != nil { |
| 54 | + annotations, ok = ann.(map[string]string) |
| 55 | + if !ok { |
| 56 | + return nil, fmt.Errorf("could not extract annotations: %v", zap.Any("annotations", ann)) |
| 57 | + } |
| 58 | + } |
| 59 | + podUID = endpointPod["uid"].(string) |
| 60 | + } else { |
| 61 | + return nil, nil |
| 62 | + } |
| 63 | + |
| 64 | + if valType, ok := env["type"]; ok { |
| 65 | + endpointType, ok = valType.(string) |
| 66 | + if !ok { |
| 67 | + return nil, fmt.Errorf("could not extract endpointType: %v", zap.Any("endpointType", valType)) |
| 68 | + } |
| 69 | + } else { |
| 70 | + return nil, fmt.Errorf("could not get endpoint type: %v", zap.Any("env", env)) |
| 71 | + } |
| 72 | + |
| 73 | + if len(annotations) > 0 { |
| 74 | + if endpointType == string(observer.PortType) && builder.config.Metrics.Enabled { |
| 75 | + // Only handle Endpoints of type port for metrics |
| 76 | + return builder.createMetricsReceiver(annotations, env, podUID) |
| 77 | + } |
| 78 | + } |
| 79 | + return nil, nil |
| 80 | +} |
| 81 | + |
| 82 | +func (builder *K8sHintsBuilder) createMetricsReceiver( |
| 83 | + annotations map[string]string, |
| 84 | + env observer.EndpointEnv, |
| 85 | + podUID string) (*receiverTemplate, error) { |
| 86 | + |
| 87 | + var port uint16 |
| 88 | + |
| 89 | + portName := env["name"].(string) |
| 90 | + subreceiverKey := getHintAnnotation(annotations, metricsHint, hintsMetricsReceiver, portName) |
| 91 | + |
| 92 | + if subreceiverKey == "" { |
| 93 | + // no metrics hints detected |
| 94 | + return nil, nil |
| 95 | + } |
| 96 | + builder.logger.Debug("handling added hinted receiver", zap.Any("subreceiverKey", subreceiverKey)) |
| 97 | + |
| 98 | + userConfMap := createMetricsConfig(annotations, env, portName) |
| 99 | + |
| 100 | + if p, ok := env["port"]; ok { |
| 101 | + port = p.(uint16) |
| 102 | + if port == 0 { |
| 103 | + return nil, fmt.Errorf("could not extract port: %v", zap.Any("env", env)) |
| 104 | + } |
| 105 | + } else { |
| 106 | + return nil, fmt.Errorf("could not extract port: %v", zap.Any("env", env)) |
| 107 | + } |
| 108 | + subreceiver, err := newReceiverTemplate(fmt.Sprintf("%v/%v_%v", subreceiverKey, podUID, port), userConfMap) |
| 109 | + if err != nil { |
| 110 | + builder.logger.Error("error adding subreceiver", zap.Any("err", err)) |
| 111 | + return nil, err |
| 112 | + } |
| 113 | + |
| 114 | + builder.logger.Debug("adding hinted receiver", zap.Any("subreceiver", subreceiver)) |
| 115 | + return &subreceiver, nil |
| 116 | + |
| 117 | +} |
| 118 | + |
| 119 | +func createMetricsConfig(annotations map[string]string, env observer.EndpointEnv, portName string) userConfigMap { |
| 120 | + confMap := map[string]any{} |
| 121 | + |
| 122 | + defaultEndpoint := env["endpoint"] |
| 123 | + // get endpoint directly from the Port endpoint |
| 124 | + if defaultEndpoint != "" { |
| 125 | + confMap["endpoint"] = defaultEndpoint |
| 126 | + } |
| 127 | + |
| 128 | + subreceiverEndpoint := getHintAnnotation(annotations, metricsHint, hintsMetricsEndpoint, portName) |
| 129 | + if subreceiverEndpoint != "" { |
| 130 | + confMap["endpoint"] = subreceiverEndpoint |
| 131 | + } |
| 132 | + subreceiverColInterval := getHintAnnotation(annotations, metricsHint, hintsMetricsCollectionInterval, portName) |
| 133 | + if subreceiverColInterval != "" { |
| 134 | + confMap["collection_interval"] = subreceiverColInterval |
| 135 | + } |
| 136 | + subreceiverTimeout := getHintAnnotation(annotations, metricsHint, hintsMetricsTimeout, portName) |
| 137 | + if subreceiverTimeout != "" { |
| 138 | + confMap["timeout"] = subreceiverTimeout |
| 139 | + } |
| 140 | + subreceiverUsername := getHintAnnotation(annotations, metricsHint, hintsMetricsUsername, portName) |
| 141 | + if subreceiverUsername != "" { |
| 142 | + confMap["username"] = subreceiverUsername |
| 143 | + } |
| 144 | + subreceiverPassword := getHintAnnotation(annotations, metricsHint, hintsMetricsPassword, portName) |
| 145 | + if subreceiverPassword != "" { |
| 146 | + confMap["password"] = subreceiverPassword |
| 147 | + } |
| 148 | + return confMap |
| 149 | +} |
| 150 | + |
| 151 | +func getHintAnnotation(annotations map[string]string, hintType string, hintKey string, suffix string) string { |
| 152 | + // try to scope the hint more on container level by suffixing with .<port_name> |
| 153 | + containerLevelHint := annotations[fmt.Sprintf("%s.%s.%s/%s", otelHints, hintType, suffix, hintKey)] |
| 154 | + if containerLevelHint != "" { |
| 155 | + return containerLevelHint |
| 156 | + } |
| 157 | + |
| 158 | + // if there is no container level hint defined try to use the Pod level hint |
| 159 | + podHintKey := fmt.Sprintf("%s.%s/%s", otelHints, hintType, hintKey) |
| 160 | + podLevelHint := annotations[podHintKey] |
| 161 | + if podLevelHint != "" { |
| 162 | + return podLevelHint |
| 163 | + } |
| 164 | + return "" |
| 165 | +} |
0 commit comments