Skip to content

Commit

Permalink
Add EMF Storage Resolution support using Metric Attributes
Browse files Browse the repository at this point in the history 
This commit addresses comments in PR#36057 where it was suggested that metric description for the EMF Exporter
should be deprecated and metric attributes used to specify the storage resolution for each metric.

The code has been modified to remove the original changes and instead have the exporter look for a metric
attribute named `aws.emf.storage_resolution` and use its value for the storage resolution when sending
a message to CloudWatch EMF.

The code has also added a new struct `cWMetricInfo` to hold the metric name, unit, and storage resolution
where previously this was a `[]map[string]string` type.

removed unnecessary changes
  • Loading branch information
@jpbarto
jpbarto committed Oct 31, 2024
1 parent 2d92ca8 commit a29de76
Show file tree
Hide file tree
Showing 9 changed files with 261 additions and 288 deletions.
8 changes: 8 additions & 0 deletions exporter/awsemfexporter/README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,14 @@ This exporter follows default credential resolution for the
Follow the [guidelines](https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html) for the
credential configuration.

## Metric Attributes
By setting attributes on your metrics you can change how individual metrics are sent to CloudWatch. Attributes can be set in code or using components like the [Attribute Processor](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/attributesprocessor).

The AWS EMF Exporter will interpret the following metric attributes to change how it publishes metrics to CloudWatch:

| Attribute Name | Description | Default |
| :---------------- | :--------------------------------------------------------------------- | ------- |
| `aws.emf.storage_resolution` | This attribute should be set to an integer value of `1` or `60`. When sending the metric value to CloudWatch use the specified storage resolution value. CloudWatch currently supports a storage resolution of `1` or `60` to indicate 1 second or 60 second resolution. | `aws.emf.storage_resolution = 60` |

## Configuration Examples

Expand Down
11 changes: 1 addition & 10 deletions exporter/awsemfexporter/config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -101,8 +101,6 @@ type MetricDescriptor struct {
// Overwrite set to true means the existing metric descriptor will be overwritten or a new metric descriptor will be created; false means
// the descriptor will only be configured if empty.
Overwrite bool `mapstructure:"overwrite"`
// Set the storage resolution for this metric in AWS Embedded Metric Format (EMF). Valid values are 1 (for 1 second resolution) and 60 (for 60 second resolution).
StorageResolution int `mapstructure:"storage_resolution"`
}

var _ component.Config = (*Config)(nil)
Expand All @@ -125,14 +123,7 @@ func (config *Config) Validate() error {
if descriptor.MetricName == "" {
continue
}
_, unitOk := eMFSupportedUnits[descriptor.Unit]
stoResErr := cwlogs.ValidateStorageResolution(descriptor.StorageResolution)

// store the descriptor if
// - it has a valid unit and a valid storage resolution
// - it has a valid unit and no specified storage resolution
// - no specified unit but a valid storage resolution
if (unitOk && stoResErr == nil) || (unitOk && descriptor.StorageResolution == 0) || (stoResErr == nil && descriptor.Unit == "") {
if _, ok := eMFSupportedUnits[descriptor.Unit]; ok {
validDescriptors = append(validDescriptors, descriptor)
} else {
config.logger.Warn("Dropped unsupported metric desctriptor.", zap.String("unit", descriptor.Unit))
Expand Down
15 changes: 4 additions & 11 deletions exporter/awsemfexporter/config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,10 +97,9 @@ func TestLoadConfig(t *testing.T) {
OutputDestination: "cloudwatch",
Version: "1",
MetricDescriptors: []MetricDescriptor{{
MetricName: "memcached_current_items",
Unit: "Count",
Overwrite: true,
StorageResolution: 1,
MetricName: "memcached_current_items",
Unit: "Count",
Overwrite: true,
}},
logger: zap.NewNop(),
},
Expand Down Expand Up @@ -128,10 +127,6 @@ func TestConfigValidate(t *testing.T) {
{Unit: "Count", MetricName: "apiserver_total", Overwrite: true},
{Unit: "INVALID", MetricName: "404"},
{Unit: "Megabytes", MetricName: "memory_usage"},
{StorageResolution: 1, MetricName: "saturation"},
{StorageResolution: 10, MetricName: "throughput"},
{Unit: "Count", MetricName: "error_total", StorageResolution: 1},
{Unit: "Count", MetricName: "error_total", StorageResolution: 5, Overwrite: true},
}
cfg := &Config{
AWSSessionSettings: awsutil.AWSSessionSettings{
Expand All @@ -145,12 +140,10 @@ func TestConfigValidate(t *testing.T) {
}
assert.NoError(t, component.ValidateConfig(cfg))

assert.Len(t, cfg.MetricDescriptors, 4)
assert.Len(t, cfg.MetricDescriptors, 2)
assert.Equal(t, []MetricDescriptor{
{Unit: "Count", MetricName: "apiserver_total", Overwrite: true},
{Unit: "Megabytes", MetricName: "memory_usage"},
{StorageResolution: 1, MetricName: "saturation"},
{Unit: "Count", MetricName: "error_total", StorageResolution: 1},
}, cfg.MetricDescriptors)
}

Expand Down
20 changes: 4 additions & 16 deletions exporter/awsemfexporter/grouped_metric.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,8 @@ type groupedMetric struct {

// metricInfo defines value and unit for OT Metrics
type metricInfo struct {
value any
unit string
storageResolution int
value any
unit string
}

// addToGroupedMetric processes OT metrics and adds them into GroupedMetric buckets
Expand Down Expand Up @@ -79,9 +78,8 @@ func addToGroupedMetric(
}

metric := &metricInfo{
value: dp.value,
unit: translateUnit(pmd, descriptor),
storageResolution: translateResolution(pmd, descriptor),
value: dp.value,
unit: translateUnit(pmd, descriptor),
}

if dp.timestampMs > 0 {
Expand Down Expand Up @@ -188,16 +186,6 @@ func mapGetHelper(labels map[string]string, key string) string {
return ""
}

func translateResolution(metric pmetric.Metric, descriptor map[string]MetricDescriptor) int {
if descriptor, exists := descriptor[metric.Name()]; exists {
if descriptor.StorageResolution == 1 || descriptor.StorageResolution == 60 {
return descriptor.StorageResolution
}
}

return 60
}

func translateUnit(metric pmetric.Metric, descriptor map[string]MetricDescriptor) string {
unit := metric.Unit()
if descriptor, exists := descriptor[metric.Name()]; exists {
Expand Down
21 changes: 8 additions & 13 deletions exporter/awsemfexporter/grouped_metric_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,8 @@ func TestAddToGroupedMetric(t *testing.T) {
expectedLabels: map[string]string{oTellibDimensionKey: instrumentationLibName, "label1": "value1"},
expectedMetricInfo: map[string]*metricInfo{
"foo": {
value: 0.1,
unit: "Count",
storageResolution: 60,
value: 0.1,
unit: "Count",
},
},
},
Expand All @@ -55,9 +54,8 @@ func TestAddToGroupedMetric(t *testing.T) {
expectedLabels: map[string]string{oTellibDimensionKey: instrumentationLibName, "label1": "value1"},
expectedMetricInfo: map[string]*metricInfo{
"foo": {
value: float64(1),
unit: "Count",
storageResolution: 60,
value: float64(1),
unit: "Count",
},
},
},
Expand All @@ -72,8 +70,7 @@ func TestAddToGroupedMetric(t *testing.T) {
Count: 18,
Sum: 35.0,
},
unit: "Seconds",
storageResolution: 60,
unit: "Seconds",
},
},
},
Expand All @@ -90,8 +87,7 @@ func TestAddToGroupedMetric(t *testing.T) {
Count: 5,
Sum: 15,
},
unit: "Seconds",
storageResolution: 60,
unit: "Seconds",
},
},
},
Expand Down Expand Up @@ -301,9 +297,8 @@ func TestAddToGroupedMetric(t *testing.T) {
assert.Len(t, group.metrics, 1)
expectedMetrics := map[string]*metricInfo{
"int-gauge": {
value: float64(1),
unit: "Count",
storageResolution: 60,
value: float64(1),
unit: "Count",
},
}
assert.Equal(t, expectedMetrics, group.metrics)
Expand Down
74 changes: 55 additions & 19 deletions exporter/awsemfexporter/metric_translator.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ import (
"encoding/json"
"fmt"
"reflect"
"regexp"
"strconv"
"time"

"go.opentelemetry.io/collector/pdata/pmetric"
Expand All @@ -29,6 +31,11 @@ const (
prometheusReceiver = "prometheus"
attributeReceiver = "receiver"
fieldPrometheusMetricType = "prom_metric_type"

// metric attributes for AWS EMF, not to be treated as metric labels
// emfAttributeFilterRE = regexp.MustCompile("^awsemf.[A-Za-z_]*$")
emfAttributeFilter = "^awsemf.[A-Za-z_]*$"
emfStorageResolutionAttribute = "aws.emf.storage_resolution"
)

var fieldPrometheusTypes = map[pmetric.MetricType]string{
Expand All @@ -45,10 +52,16 @@ type cWMetrics struct {
fields map[string]any
}

type cWMetricInfo struct {
Name string
Unit string
StorageResolution int
}

type cWMeasurement struct {
Namespace string
Dimensions [][]string
Metrics []map[string]any
Metrics []cWMetricInfo
}

type cWMetricStats struct {
Expand Down Expand Up @@ -156,7 +169,7 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri

// translateGroupedMetricToCWMetric converts Grouped Metric format to CloudWatch Metric format.
func translateGroupedMetricToCWMetric(groupedMetric *groupedMetric, config *Config) *cWMetrics {
labels := groupedMetric.labels
labels := filterAWSEMFAttributes(groupedMetric.labels)
fieldsLength := len(labels) + len(groupedMetric.metrics)

isPrometheusMetric := groupedMetric.metadata.receiver == prometheusReceiver
Expand Down Expand Up @@ -198,7 +211,7 @@ func translateGroupedMetricToCWMetric(groupedMetric *groupedMetric, config *Conf

// groupedMetricToCWMeasurement creates a single CW Measurement from a grouped metric.
func groupedMetricToCWMeasurement(groupedMetric *groupedMetric, config *Config) cWMeasurement {
labels := groupedMetric.labels
labels := filterAWSEMFAttributes(groupedMetric.labels)
dimensionRollupOption := config.DimensionRollupOption

// Create a dimension set containing list of label names
Expand All @@ -208,8 +221,13 @@ func groupedMetricToCWMeasurement(groupedMetric *groupedMetric, config *Config)
dimSet[idx] = labelName
idx++
}

fmt.Printf("##--## Dim Set: %+v\n", dimSet)

dimensions := [][]string{dimSet}

fmt.Printf("##--## Dimensions: %+v", dimensions)

// Apply single/zero dimension rollup to labels
rollupDimensionArray := dimensionRollup(dimensionRollupOption, labels)

Expand All @@ -228,18 +246,22 @@ func groupedMetricToCWMeasurement(groupedMetric *groupedMetric, config *Config)
// Add on rolled-up dimensions
dimensions = append(dimensions, rollupDimensionArray...)

metrics := make([]map[string]any, len(groupedMetric.metrics))
fmt.Printf("##--## Post Dimensions: %+v", dimensions)

metrics := make([]cWMetricInfo, len(groupedMetric.metrics))
idx = 0
for metricName, metricInfo := range groupedMetric.metrics {
metrics[idx] = map[string]any{
"Name": metricName,
"StorageResolution": 60,
metrics[idx] = cWMetricInfo{
Name: metricName,
StorageResolution: 60,
}
if metricInfo.unit != "" {
metrics[idx]["Unit"] = metricInfo.unit
metrics[idx].Unit = metricInfo.unit
}
if resErr := cwlogs.ValidateStorageResolution(metricInfo.storageResolution); resErr == nil {
metrics[idx]["StorageResolution"] = metricInfo.storageResolution
if storRes, ok := groupedMetric.labels[emfStorageResolutionAttribute]; ok {
if storResInt, err := strconv.Atoi(storRes); err == nil {
metrics[idx].StorageResolution = storResInt
}
}
idx++
}
Expand All @@ -254,7 +276,7 @@ func groupedMetricToCWMeasurement(groupedMetric *groupedMetric, config *Config)
// groupedMetricToCWMeasurementsWithFilters filters the grouped metric using the given list of metric
// declarations and returns the corresponding list of CW Measurements.
func groupedMetricToCWMeasurementsWithFilters(groupedMetric *groupedMetric, config *Config) (cWMeasurements []cWMeasurement) {
labels := groupedMetric.labels
labels := filterAWSEMFAttributes(groupedMetric.labels)

// Filter metric declarations by labels
metricDeclarations := make([]*MetricDeclaration, 0, len(config.MetricDeclarations))
Expand Down Expand Up @@ -282,7 +304,7 @@ func groupedMetricToCWMeasurementsWithFilters(groupedMetric *groupedMetric, conf
// Group metrics by matched metric declarations
type metricDeclarationGroup struct {
metricDeclIdxList []int
metrics []map[string]any
metrics []cWMetricInfo
}

metricDeclGroups := make(map[string]*metricDeclarationGroup)
Expand All @@ -303,23 +325,25 @@ func groupedMetricToCWMeasurementsWithFilters(groupedMetric *groupedMetric, conf
continue
}

metric := map[string]any{
"Name": metricName,
"StorageResolution": 60,
metric := cWMetricInfo{
Name: metricName,
StorageResolution: 60,
}
if metricInfo.unit != "" {
metric["Unit"] = metricInfo.unit
metric.Unit = metricInfo.unit
}
if resErr := cwlogs.ValidateStorageResolution(metricInfo.storageResolution); resErr == nil {
metric["StorageResolution"] = metricInfo.storageResolution
if storRes, ok := groupedMetric.labels[emfStorageResolutionAttribute]; ok {
if storResInt, err := strconv.Atoi(storRes); err == nil {
metric.StorageResolution = storResInt
}
}
metricDeclKey := fmt.Sprint(metricDeclIdx)
if group, ok := metricDeclGroups[metricDeclKey]; ok {
group.metrics = append(group.metrics, metric)
} else {
metricDeclGroups[metricDeclKey] = &metricDeclarationGroup{
metricDeclIdxList: metricDeclIdx,
metrics: []map[string]any{metric},
metrics: []cWMetricInfo{metric},
}
}
}
Expand Down Expand Up @@ -475,3 +499,15 @@ func translateGroupedMetricToEmf(groupedMetric *groupedMetric, config *Config, d

return event, nil
}

func filterAWSEMFAttributes(labels map[string]string) map[string]string {
// remove any labels that are attributes specific to AWS EMF Exporter
filteredLabels := make(map[string]string)
emfAttributeFilterRE := regexp.MustCompile(emfAttributeFilter)
for labelName := range labels {
if !emfAttributeFilterRE.MatchString(labelName) {
filteredLabels[labelName] = labels[labelName]
}
}
return filteredLabels
}
Loading

0 comments on commit a29de76

Please sign in to comment.