From 542699b185e85bce397a7227c909b40023fc4011 Mon Sep 17 00:00:00 2001 From: Dominik Rosiek Date: Tue, 9 Apr 2024 16:53:00 +0200 Subject: [PATCH 1/6] refactor!: remove compress.go Signed-off-by: Dominik Rosiek --- exporter/sumologicexporter/compress.go | 77 ----------- exporter/sumologicexporter/compress_test.go | 142 -------------------- 2 files changed, 219 deletions(-) delete mode 100644 exporter/sumologicexporter/compress.go delete mode 100644 exporter/sumologicexporter/compress_test.go diff --git a/exporter/sumologicexporter/compress.go b/exporter/sumologicexporter/compress.go deleted file mode 100644 index e18df4bc9286..000000000000 --- a/exporter/sumologicexporter/compress.go +++ /dev/null @@ -1,77 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" - -import ( - "bytes" - "compress/flate" - "compress/gzip" - "fmt" - "io" -) - -type compressor struct { - format CompressEncodingType - writer encoder - buf bytes.Buffer -} - -type encoder interface { - io.WriteCloser - Reset(dst io.Writer) -} - -// newCompressor takes encoding format and returns compressor struct and error eventually -func newCompressor(format CompressEncodingType) (compressor, error) { - var ( - writer encoder - err error - ) - - switch format { - case GZIPCompression: - writer = gzip.NewWriter(io.Discard) - case DeflateCompression: - writer, err = flate.NewWriter(io.Discard, flate.BestSpeed) - if err != nil { - return compressor{}, err - } - case NoCompression: - writer = nil - default: - return compressor{}, fmt.Errorf("invalid format: %s", format) - } - - return compressor{ - format: format, - writer: writer, - }, nil -} - -// compress takes a reader with uncompressed data and returns -// a reader with the same data compressed using c.writer -func (c *compressor) compress(data io.Reader) (io.Reader, error) { - if c.writer == nil { - return data, nil - } - - var dataBytes bytes.Buffer - if _, err := dataBytes.ReadFrom(data); err != nil { - return nil, err - } - - // Reset c.buf to start with empty message - c.buf.Reset() - c.writer.Reset(&c.buf) - - if _, err := c.writer.Write(dataBytes.Bytes()); err != nil { - return nil, err - } - - if err := c.writer.Close(); err != nil { - return nil, err - } - - return bytes.NewReader(c.buf.Bytes()), nil -} diff --git a/exporter/sumologicexporter/compress_test.go b/exporter/sumologicexporter/compress_test.go deleted file mode 100644 index 0300e4f7cc5d..000000000000 --- a/exporter/sumologicexporter/compress_test.go +++ /dev/null @@ -1,142 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter - -import ( - "compress/flate" - "compress/gzip" - "errors" - "io" - "strings" - "testing" - - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -type mockedEncrypter struct { - writeError error - closeError error -} - -func (e mockedEncrypter) Reset(_ io.Writer) { -} - -func (e mockedEncrypter) Write(_ []byte) (n int, err error) { - return 0, e.writeError -} - -func (e mockedEncrypter) Close() error { - return e.closeError -} - -func getTestCompressor(w error, c error) compressor { - return compressor{ - format: GZIPCompression, - writer: mockedEncrypter{ - writeError: w, - closeError: c, - }, - } -} - -type mockedReader struct{} - -func (r mockedReader) Read(_ []byte) (n int, err error) { - return 0, errors.New("read error") -} - -func TestCompressGzip(t *testing.T) { - const message = "This is an example log" - - c, err := newCompressor(GZIPCompression) - require.NoError(t, err) - - body := strings.NewReader(message) - - data, err := c.compress(body) - require.NoError(t, err) - - assert.Equal(t, message, decodeGzip(t, data)) -} - -func TestCompressTwice(t *testing.T) { - const ( - message = "This is an example log" - secondMessage = "This is an another example log" - ) - - c, err := newCompressor(GZIPCompression) - require.NoError(t, err) - - body := strings.NewReader(message) - - data, err := c.compress(body) - require.NoError(t, err) - assert.Equal(t, message, decodeGzip(t, data)) - - body = strings.NewReader(secondMessage) - data, err = c.compress(body) - require.NoError(t, err) - assert.Equal(t, secondMessage, decodeGzip(t, data)) -} - -func decodeGzip(t *testing.T, data io.Reader) string { - r, err := gzip.NewReader(data) - require.NoError(t, err) - - var buf []byte - buf, err = io.ReadAll(r) - require.NoError(t, err) - - return string(buf) -} - -func TestCompressDeflate(t *testing.T) { - const message = "This is an example log" - - c, err := newCompressor(DeflateCompression) - require.NoError(t, err) - - body := strings.NewReader(message) - - data, err := c.compress(body) - require.NoError(t, err) - - assert.Equal(t, message, decodeDeflate(t, data)) -} - -func decodeDeflate(t *testing.T, data io.Reader) string { - r := flate.NewReader(data) - - var buf []byte - buf, err := io.ReadAll(r) - require.NoError(t, err) - - return string(buf) -} - -func TestCompressReadError(t *testing.T) { - c := getTestCompressor(nil, nil) - r := mockedReader{} - _, err := c.compress(r) - - assert.EqualError(t, err, "read error") -} - -func TestCompressWriteError(t *testing.T) { - c := getTestCompressor(errors.New("write error"), nil) - r := strings.NewReader("test string") - _, err := c.compress(r) - - assert.EqualError(t, err, "write error") -} - -func TestCompressCloseError(t *testing.T) { - c := getTestCompressor(nil, errors.New("close error")) - r := strings.NewReader("test string") - _, err := c.compress(r) - - assert.EqualError(t, err, "close error") -} From a323e5ca3ed39f175004ee67f385235381a11eb3 Mon Sep 17 00:00:00 2001 From: Dominik Rosiek Date: Tue, 9 Apr 2024 16:57:03 +0200 Subject: [PATCH 2/6] refactor!: graphite and carbon Signed-off-by: Dominik Rosiek --- .../sumologicexporter/carbon_formatter.go | 103 ------------- .../carbon_formatter_test.go | 88 ----------- .../sumologicexporter/graphite_formatter.go | 115 --------------- .../graphite_formatter_test.go | 137 ------------------ 4 files changed, 443 deletions(-) delete mode 100644 exporter/sumologicexporter/carbon_formatter.go delete mode 100644 exporter/sumologicexporter/carbon_formatter_test.go delete mode 100644 exporter/sumologicexporter/graphite_formatter.go delete mode 100644 exporter/sumologicexporter/graphite_formatter_test.go diff --git a/exporter/sumologicexporter/carbon_formatter.go b/exporter/sumologicexporter/carbon_formatter.go deleted file mode 100644 index aed80272325c..000000000000 --- a/exporter/sumologicexporter/carbon_formatter.go +++ /dev/null @@ -1,103 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" - -import ( - "fmt" - "strings" - - "go.opentelemetry.io/collector/pdata/pcommon" - "go.opentelemetry.io/collector/pdata/pmetric" -) - -// carbon2TagString returns all attributes as space spearated key=value pairs. -// In addition, metric name and unit are also included. -// In case `metric` or `unit` attributes has been set too, they are prefixed -// with underscore `_` to avoid overwriting the metric name and unit. -func carbon2TagString(record metricPair) string { - length := record.attributes.Len() - - if _, ok := record.attributes.Get("metric"); ok { - length++ - } - - if _, ok := record.attributes.Get("unit"); ok && len(record.metric.Unit()) > 0 { - length++ - } - - returnValue := make([]string, 0, length) - record.attributes.Range(func(k string, v pcommon.Value) bool { - if k == "name" || k == "unit" { - k = fmt.Sprintf("_%s", k) - } - returnValue = append(returnValue, fmt.Sprintf( - "%s=%s", - sanitizeCarbonString(k), - sanitizeCarbonString(v.AsString()), - )) - return true - }) - - returnValue = append(returnValue, fmt.Sprintf("metric=%s", sanitizeCarbonString(record.metric.Name()))) - - if len(record.metric.Unit()) > 0 { - returnValue = append(returnValue, fmt.Sprintf("unit=%s", sanitizeCarbonString(record.metric.Unit()))) - } - - return strings.Join(returnValue, " ") -} - -// sanitizeCarbonString replaces problematic characters with underscore -func sanitizeCarbonString(text string) string { - return strings.NewReplacer(" ", "_", "=", ":", "\n", "_").Replace(text) -} - -// carbon2NumberRecord converts NumberDataPoint to carbon2 metric string -// with additional information from metricPair. -func carbon2NumberRecord(record metricPair, dataPoint pmetric.NumberDataPoint) string { - switch dataPoint.ValueType() { - case pmetric.NumberDataPointValueTypeDouble: - return fmt.Sprintf("%s %g %d", - carbon2TagString(record), - dataPoint.DoubleValue(), - dataPoint.Timestamp()/1e9, - ) - case pmetric.NumberDataPointValueTypeInt: - return fmt.Sprintf("%s %d %d", - carbon2TagString(record), - dataPoint.IntValue(), - dataPoint.Timestamp()/1e9, - ) - case pmetric.NumberDataPointValueTypeEmpty: - return "" - } - return "" -} - -// carbon2metric2String converts metric to Carbon2 formatted string. -func carbon2Metric2String(record metricPair) string { - var nextLines []string - //exhaustive:enforce - switch record.metric.Type() { - case pmetric.MetricTypeGauge: - dps := record.metric.Gauge().DataPoints() - nextLines = make([]string, 0, dps.Len()) - for i := 0; i < dps.Len(); i++ { - nextLines = append(nextLines, carbon2NumberRecord(record, dps.At(i))) - } - case pmetric.MetricTypeSum: - dps := record.metric.Sum().DataPoints() - nextLines = make([]string, 0, dps.Len()) - for i := 0; i < dps.Len(); i++ { - nextLines = append(nextLines, carbon2NumberRecord(record, dps.At(i))) - } - // Skip complex metrics - case pmetric.MetricTypeHistogram: - case pmetric.MetricTypeSummary: - case pmetric.MetricTypeEmpty: - case pmetric.MetricTypeExponentialHistogram: - } - - return strings.Join(nextLines, "\n") -} diff --git a/exporter/sumologicexporter/carbon_formatter_test.go b/exporter/sumologicexporter/carbon_formatter_test.go deleted file mode 100644 index 5f361721d0c9..000000000000 --- a/exporter/sumologicexporter/carbon_formatter_test.go +++ /dev/null @@ -1,88 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter - -import ( - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestCarbon2TagString(t *testing.T) { - metric := exampleIntMetric() - data := carbon2TagString(metric) - assert.Equal(t, "test=test_value test2=second_value metric=test.metric.data unit=bytes", data) - - metric = exampleIntGaugeMetric() - data = carbon2TagString(metric) - assert.Equal(t, "foo=bar metric=gauge_metric_name", data) - - metric = exampleDoubleSumMetric() - data = carbon2TagString(metric) - assert.Equal(t, "foo=bar metric=sum_metric_double_test", data) - - metric = exampleDoubleGaugeMetric() - data = carbon2TagString(metric) - assert.Equal(t, "foo=bar metric=gauge_metric_name_double_test", data) -} - -func TestCarbonMetricTypeIntGauge(t *testing.T) { - metric := exampleIntGaugeMetric() - - result := carbon2Metric2String(metric) - expected := `foo=bar metric=gauge_metric_name 124 1608124661 -foo=bar metric=gauge_metric_name 245 1608124662` - assert.Equal(t, expected, result) -} - -func TestCarbonMetricTypeDoubleGauge(t *testing.T) { - metric := exampleDoubleGaugeMetric() - - result := carbon2Metric2String(metric) - expected := `foo=bar metric=gauge_metric_name_double_test 33.4 1608124661 -foo=bar metric=gauge_metric_name_double_test 56.8 1608124662` - assert.Equal(t, expected, result) -} - -func TestCarbonMetricTypeIntSum(t *testing.T) { - metric := exampleIntSumMetric() - - result := carbon2Metric2String(metric) - expected := `foo=bar metric=sum_metric_int_test 45 1608124444 -foo=bar metric=sum_metric_int_test 1238 1608124699` - assert.Equal(t, expected, result) -} - -func TestCarbonMetricTypeDoubleSum(t *testing.T) { - metric := exampleDoubleSumMetric() - - result := carbon2Metric2String(metric) - expected := `foo=bar metric=sum_metric_double_test 45.6 1618124444 -foo=bar metric=sum_metric_double_test 1238.1 1608424699` - assert.Equal(t, expected, result) -} - -func TestCarbonMetricTypeSummary(t *testing.T) { - metric := exampleSummaryMetric() - - result := carbon2Metric2String(metric) - expected := `` - assert.Equal(t, expected, result) - - metric = buildExampleSummaryMetric(false) - result = carbon2Metric2String(metric) - assert.Equal(t, expected, result) -} - -func TestCarbonMetricTypeHistogram(t *testing.T) { - metric := exampleHistogramMetric() - - result := carbon2Metric2String(metric) - expected := `` - assert.Equal(t, expected, result) - - metric = buildExampleHistogramMetric(false) - result = carbon2Metric2String(metric) - assert.Equal(t, expected, result) -} diff --git a/exporter/sumologicexporter/graphite_formatter.go b/exporter/sumologicexporter/graphite_formatter.go deleted file mode 100644 index 3d3c605834ef..000000000000 --- a/exporter/sumologicexporter/graphite_formatter.go +++ /dev/null @@ -1,115 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" - -import ( - "fmt" - "regexp" - "strings" - "time" - - "go.opentelemetry.io/collector/pdata/pcommon" - "go.opentelemetry.io/collector/pdata/pmetric" -) - -type graphiteFormatter struct { - template sourceFormat - replacer *strings.Replacer -} - -const ( - graphiteMetricNamePlaceholder = "_metric_" -) - -// newGraphiteFormatter creates new formatter for given SourceFormat template -func newGraphiteFormatter(template string) graphiteFormatter { - r := regexp.MustCompile(sourceRegex) - - sf := newSourceFormat(r, template) - - return graphiteFormatter{ - template: sf, - replacer: strings.NewReplacer(`.`, `_`, ` `, `_`), - } -} - -// escapeGraphiteString replaces dot and space using replacer, -// as dot is special character for graphite format -func (gf *graphiteFormatter) escapeGraphiteString(value string) string { - return gf.replacer.Replace(value) -} - -// format returns metric name basing on template for given fields nas metric name -func (gf *graphiteFormatter) format(f fields, metricName string) string { - s := gf.template - labels := make([]any, 0, len(s.matches)) - - for _, matchset := range s.matches { - if matchset == graphiteMetricNamePlaceholder { - labels = append(labels, gf.escapeGraphiteString(metricName)) - } else { - attr, ok := f.orig.Get(matchset) - var value string - if ok { - value = attr.AsString() - } else { - value = "" - } - labels = append(labels, gf.escapeGraphiteString(value)) - } - } - - return fmt.Sprintf(s.template, labels...) -} - -// numberRecord converts NumberDataPoint to graphite metric string -// with additional information from fields -func (gf *graphiteFormatter) numberRecord(fs fields, name string, dataPoint pmetric.NumberDataPoint) string { - switch dataPoint.ValueType() { - case pmetric.NumberDataPointValueTypeDouble: - return fmt.Sprintf("%s %g %d", - gf.format(fs, name), - dataPoint.DoubleValue(), - dataPoint.Timestamp()/pcommon.Timestamp(time.Second), - ) - case pmetric.NumberDataPointValueTypeInt: - return fmt.Sprintf("%s %d %d", - gf.format(fs, name), - dataPoint.IntValue(), - dataPoint.Timestamp()/pcommon.Timestamp(time.Second), - ) - case pmetric.NumberDataPointValueTypeEmpty: - return "" - } - return "" -} - -// metric2String returns stringified metricPair -func (gf *graphiteFormatter) metric2String(record metricPair) string { - var nextLines []string - fs := newFields(record.attributes) - name := record.metric.Name() - //exhaustive:enforce - switch record.metric.Type() { - case pmetric.MetricTypeGauge: - dps := record.metric.Gauge().DataPoints() - nextLines = make([]string, 0, dps.Len()) - for i := 0; i < dps.Len(); i++ { - nextLines = append(nextLines, gf.numberRecord(fs, name, dps.At(i))) - } - case pmetric.MetricTypeSum: - dps := record.metric.Sum().DataPoints() - nextLines = make([]string, 0, dps.Len()) - for i := 0; i < dps.Len(); i++ { - nextLines = append(nextLines, gf.numberRecord(fs, name, dps.At(i))) - } - // Skip complex metrics - case pmetric.MetricTypeHistogram: - case pmetric.MetricTypeSummary: - case pmetric.MetricTypeEmpty: - case pmetric.MetricTypeExponentialHistogram: - } - - return strings.Join(nextLines, "\n") -} diff --git a/exporter/sumologicexporter/graphite_formatter_test.go b/exporter/sumologicexporter/graphite_formatter_test.go deleted file mode 100644 index 0b2d27971415..000000000000 --- a/exporter/sumologicexporter/graphite_formatter_test.go +++ /dev/null @@ -1,137 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter - -import ( - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestEscapeGraphiteString(t *testing.T) { - gf := newGraphiteFormatter("%{k8s.cluster}.%{k8s.namespace}.%{k8s.pod}.%{_metric_}") - - value := gf.escapeGraphiteString("this.is_example&metric.value") - expected := "this_is_example&metric_value" - - assert.Equal(t, expected, value) -} - -func TestGraphiteFormat(t *testing.T) { - gf := newGraphiteFormatter("%{k8s.cluster}.%{k8s.namespace}.%{k8s.pod}.%{_metric_}") - - fs := fieldsFromMap(map[string]string{ - "k8s.cluster": "test_cluster", - "k8s.namespace": "sumologic", - "k8s.pod": "example_pod", - }) - - expected := "test_cluster.sumologic.example_pod.test_metric" - result := gf.format(fs, "test_metric") - - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeIntGauge(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleIntGaugeMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `my_cluster.default.some_pod.gauge_metric_name 124 1608124661 -my_cluster.default.some_pod.gauge_metric_name 245 1608124662` - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeDoubleGauge(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleDoubleGaugeMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `my_cluster.default.some_pod.gauge_metric_name_double_test 33.4 1608124661 -my_cluster.default.some_pod.gauge_metric_name_double_test 56.8 1608124662` - assert.Equal(t, expected, result) -} - -func TestGraphiteNoattribute(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleDoubleGaugeMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `my_cluster..some_pod.gauge_metric_name_double_test 33.4 1608124661 -my_cluster..some_pod.gauge_metric_name_double_test 56.8 1608124662` - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeIntSum(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleIntSumMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `my_cluster.default.some_pod.sum_metric_int_test 45 1608124444 -my_cluster.default.some_pod.sum_metric_int_test 1238 1608124699` - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeDoubleSum(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleDoubleSumMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `my_cluster.default.some_pod.sum_metric_double_test 45.6 1618124444 -my_cluster.default.some_pod.sum_metric_double_test 1238.1 1608424699` - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeSummary(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleSummaryMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `` - assert.Equal(t, expected, result) - - metric = buildExampleSummaryMetric(false) - result = gf.metric2String(metric) - assert.Equal(t, expected, result) -} - -func TestGraphiteMetricTypeHistogram(t *testing.T) { - gf := newGraphiteFormatter("%{cluster}.%{namespace}.%{pod}.%{_metric_}") - - metric := exampleHistogramMetric() - metric.attributes.PutStr("cluster", "my_cluster") - metric.attributes.PutStr("namespace", "default") - metric.attributes.PutStr("pod", "some pod") - - result := gf.metric2String(metric) - expected := `` - assert.Equal(t, expected, result) - - metric = buildExampleHistogramMetric(false) - result = gf.metric2String(metric) - assert.Equal(t, expected, result) -} From f148392ce27bf1335235584b3722b9aa6dab84cd Mon Sep 17 00:00:00 2001 From: Dominik Rosiek Date: Tue, 9 Apr 2024 17:06:48 +0200 Subject: [PATCH 3/6] refactor!: clean remaining files Signed-off-by: Dominik Rosiek --- exporter/sumologicexporter/source_format.go | 77 ----- .../sumologicexporter/source_format_test.go | 97 ------- exporter/sumologicexporter/test_data.go | 266 ------------------ 3 files changed, 440 deletions(-) delete mode 100644 exporter/sumologicexporter/source_format.go delete mode 100644 exporter/sumologicexporter/source_format_test.go delete mode 100644 exporter/sumologicexporter/test_data.go diff --git a/exporter/sumologicexporter/source_format.go b/exporter/sumologicexporter/source_format.go deleted file mode 100644 index 3591c3be319e..000000000000 --- a/exporter/sumologicexporter/source_format.go +++ /dev/null @@ -1,77 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" - -import ( - "fmt" - "regexp" -) - -type sourceFormats struct { - name sourceFormat - host sourceFormat - category sourceFormat -} - -type sourceFormat struct { - matches []string - template string -} - -const sourceRegex = `\%\{([\w\.]+)\}` - -// newSourceFormat builds sourceFormat basing on the regex and given text. -// Regex is basing on the `sourceRegex` const -// For given example text: `%{cluster}/%{namespace}“, it sets: -// - template to `%s/%s`, which can be used later by fmt.Sprintf -// - matches as map of (attribute) keys ({"cluster", "namespace"}) which will -// be used to put corresponding value into templates' `%s -func newSourceFormat(r *regexp.Regexp, text string) sourceFormat { - matches := r.FindAllStringSubmatch(text, -1) - template := r.ReplaceAllString(text, "%s") - - m := make([]string, len(matches)) - - for i, match := range matches { - m[i] = match[1] - } - - return sourceFormat{ - matches: m, - template: template, - } -} - -// newSourceFormats returns sourceFormats for name, host and category based on cfg -func newSourceFormats(cfg *Config) sourceFormats { - r := regexp.MustCompile(sourceRegex) - - return sourceFormats{ - category: newSourceFormat(r, cfg.SourceCategory), - host: newSourceFormat(r, cfg.SourceHost), - name: newSourceFormat(r, cfg.SourceName), - } -} - -// format converts sourceFormat to string. -// Takes fields and put into template (%s placeholders) in order defined by matches -func (s *sourceFormat) format(f fields) string { - labels := make([]any, 0, len(s.matches)) - - for _, matchset := range s.matches { - v, ok := f.orig.Get(matchset) - if ok { - labels = append(labels, v.AsString()) - } else { - labels = append(labels, "") - } - } - - return fmt.Sprintf(s.template, labels...) -} - -// isSet returns true if template is non-empty -func (s *sourceFormat) isSet() bool { - return len(s.template) > 0 -} diff --git a/exporter/sumologicexporter/source_format_test.go b/exporter/sumologicexporter/source_format_test.go deleted file mode 100644 index 86316972d77d..000000000000 --- a/exporter/sumologicexporter/source_format_test.go +++ /dev/null @@ -1,97 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter - -import ( - "regexp" - "testing" - - "github.com/stretchr/testify/assert" -) - -func getTestSourceFormat(template string) sourceFormat { - r := regexp.MustCompile(sourceRegex) - - return newSourceFormat(r, template) -} - -func TestNewSourceFormat(t *testing.T) { - expected := sourceFormat{ - matches: []string{ - "test", - }, - template: "%s/test", - } - - r := regexp.MustCompile(sourceRegex) - - s := newSourceFormat(r, "%{test}/test") - - assert.Equal(t, expected, s) -} - -func TestNewSourceFormats(t *testing.T) { - expected := sourceFormats{ - host: sourceFormat{ - matches: []string{ - "namespace", - }, - template: "ns/%s", - }, - name: sourceFormat{ - matches: []string{ - "pod", - }, - template: "name/%s", - }, - category: sourceFormat{ - matches: []string{ - "cluster", - }, - template: "category/%s", - }, - } - - cfg := &Config{ - SourceName: "name/%{pod}", - SourceHost: "ns/%{namespace}", - SourceCategory: "category/%{cluster}", - } - - s := newSourceFormats(cfg) - - assert.Equal(t, expected, s) -} - -func TestFormat(t *testing.T) { - f := fieldsFromMap(map[string]string{ - "key_1": "value_1", - "key_2.subkey": "value_2", - }) - s := getTestSourceFormat("%{key_1}/%{key_2.subkey}") - expected := "value_1/value_2" - - result := s.format(f) - assert.Equal(t, expected, result) -} - -func TestFormatNonExistingKey(t *testing.T) { - f := fieldsFromMap(map[string]string{"key_2": "value_2"}) - s := getTestSourceFormat("%{key_1}/%{key_2}") - - expected := "/value_2" - - result := s.format(f) - assert.Equal(t, expected, result) -} - -func TestIsSet(t *testing.T) { - s := getTestSourceFormat("%{key_1}/%{key_2}") - assert.True(t, s.isSet()) -} - -func TestIsNotSet(t *testing.T) { - s := getTestSourceFormat("") - assert.False(t, s.isSet()) -} diff --git a/exporter/sumologicexporter/test_data.go b/exporter/sumologicexporter/test_data.go deleted file mode 100644 index 320ef4f0af83..000000000000 --- a/exporter/sumologicexporter/test_data.go +++ /dev/null @@ -1,266 +0,0 @@ -// Copyright The OpenTelemetry Authors -// SPDX-License-Identifier: Apache-2.0 - -package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" - -import ( - "go.opentelemetry.io/collector/pdata/pcommon" - "go.opentelemetry.io/collector/pdata/pmetric" -) - -func exampleIntMetric() metricPair { - return buildExampleIntMetric(true) -} - -func buildExampleIntMetric(fillData bool) metricPair { - metric := pmetric.NewMetric() - metric.SetName("test.metric.data") - metric.SetUnit("bytes") - metric.SetEmptySum() - - if fillData { - dp := metric.Sum().DataPoints().AppendEmpty() - dp.SetTimestamp(1605534165 * 1e9) - dp.SetIntValue(14500) - } - - attributes := pcommon.NewMap() - attributes.PutStr("test", "test_value") - attributes.PutStr("test2", "second_value") - - return metricPair{ - metric: metric, - attributes: attributes, - } -} - -func exampleIntGaugeMetric() metricPair { - return buildExampleIntGaugeMetric(true) -} - -func buildExampleIntGaugeMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetName("gauge_metric_name") - metric.metric.SetEmptyGauge() - - metric.attributes.PutStr("foo", "bar") - - if fillData { - dp := metric.metric.Gauge().DataPoints().AppendEmpty() - dp.Attributes().PutStr("remote_name", "156920") - dp.Attributes().PutStr("url", "http://example_url") - dp.SetIntValue(124) - dp.SetTimestamp(1608124661.166 * 1e9) - - dp = metric.metric.Gauge().DataPoints().AppendEmpty() - dp.Attributes().PutStr("remote_name", "156955") - dp.Attributes().PutStr("url", "http://another_url") - dp.SetIntValue(245) - dp.SetTimestamp(1608124662.166 * 1e9) - } - - return metric -} - -func exampleDoubleGaugeMetric() metricPair { - return buildExampleDoubleGaugeMetric(true) -} - -func buildExampleDoubleGaugeMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetEmptyGauge() - metric.metric.SetName("gauge_metric_name_double_test") - - metric.attributes.PutStr("foo", "bar") - - if fillData { - dp := metric.metric.Gauge().DataPoints().AppendEmpty() - dp.Attributes().PutStr("local_name", "156720") - dp.Attributes().PutStr("endpoint", "http://example_url") - dp.SetDoubleValue(33.4) - dp.SetTimestamp(1608124661.169 * 1e9) - - dp = metric.metric.Gauge().DataPoints().AppendEmpty() - dp.Attributes().PutStr("local_name", "156155") - dp.Attributes().PutStr("endpoint", "http://another_url") - dp.SetDoubleValue(56.8) - dp.SetTimestamp(1608124662.186 * 1e9) - } - - return metric -} - -func exampleIntSumMetric() metricPair { - return buildExampleIntSumMetric(true) -} - -func buildExampleIntSumMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetEmptySum() - metric.metric.SetName("sum_metric_int_test") - - metric.attributes.PutStr("foo", "bar") - - if fillData { - dp := metric.metric.Sum().DataPoints().AppendEmpty() - dp.Attributes().PutStr("name", "156720") - dp.Attributes().PutStr("address", "http://example_url") - dp.SetIntValue(45) - dp.SetTimestamp(1608124444.169 * 1e9) - - dp = metric.metric.Sum().DataPoints().AppendEmpty() - dp.Attributes().PutStr("name", "156155") - dp.Attributes().PutStr("address", "http://another_url") - dp.SetIntValue(1238) - dp.SetTimestamp(1608124699.186 * 1e9) - } - - return metric -} - -func exampleDoubleSumMetric() metricPair { - return buildExampleDoubleSumMetric(true) -} - -func buildExampleDoubleSumMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetEmptySum() - metric.metric.SetName("sum_metric_double_test") - - metric.attributes.PutStr("foo", "bar") - - if fillData { - dp := metric.metric.Sum().DataPoints().AppendEmpty() - dp.Attributes().PutStr("pod_name", "lorem") - dp.Attributes().PutStr("namespace", "default") - dp.SetDoubleValue(45.6) - dp.SetTimestamp(1618124444.169 * 1e9) - - dp = metric.metric.Sum().DataPoints().AppendEmpty() - dp.Attributes().PutStr("pod_name", "opsum") - dp.Attributes().PutStr("namespace", "kube-config") - dp.SetDoubleValue(1238.1) - dp.SetTimestamp(1608424699.186 * 1e9) - } - - return metric -} - -func exampleSummaryMetric() metricPair { - return buildExampleSummaryMetric(true) -} - -func buildExampleSummaryMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetEmptySummary() - metric.metric.SetName("summary_metric_double_test") - - metric.attributes.PutStr("foo", "bar") - - if fillData { - dp := metric.metric.Summary().DataPoints().AppendEmpty() - dp.Attributes().PutStr("pod_name", "dolor") - dp.Attributes().PutStr("namespace", "sumologic") - dp.SetSum(45.6) - dp.SetCount(3) - dp.SetTimestamp(1618124444.169 * 1e9) - - quantile := dp.QuantileValues().AppendEmpty() - quantile.SetQuantile(0.6) - quantile.SetValue(0.7) - - quantile = dp.QuantileValues().AppendEmpty() - quantile.SetQuantile(2.6) - quantile.SetValue(4) - - dp = metric.metric.Summary().DataPoints().AppendEmpty() - dp.Attributes().PutStr("pod_name", "sit") - dp.Attributes().PutStr("namespace", "main") - dp.SetSum(1238.1) - dp.SetCount(7) - dp.SetTimestamp(1608424699.186 * 1e9) - } - - return metric -} - -func exampleHistogramMetric() metricPair { - return buildExampleHistogramMetric(true) -} - -func buildExampleHistogramMetric(fillData bool) metricPair { - metric := metricPair{ - attributes: pcommon.NewMap(), - metric: pmetric.NewMetric(), - } - - metric.metric.SetEmptyHistogram() - metric.metric.SetName("histogram_metric_double_test") - - metric.attributes.PutStr("bar", "foo") - - if fillData { - dp := metric.metric.Histogram().DataPoints().AppendEmpty() - dp.Attributes().PutStr("container", "dolor") - dp.Attributes().PutStr("branch", "sumologic") - dp.BucketCounts().FromRaw([]uint64{0, 12, 7, 5, 8, 13}) - dp.ExplicitBounds().FromRaw([]float64{0.1, 0.2, 0.5, 0.8, 1}) - dp.SetTimestamp(1618124444.169 * 1e9) - dp.SetSum(45.6) - dp.SetCount(7) - - dp = metric.metric.Histogram().DataPoints().AppendEmpty() - dp.Attributes().PutStr("container", "sit") - dp.Attributes().PutStr("branch", "main") - dp.BucketCounts().FromRaw([]uint64{0, 10, 1, 1, 4, 6}) - dp.ExplicitBounds().FromRaw([]float64{0.1, 0.2, 0.5, 0.8, 1}) - dp.SetTimestamp(1608424699.186 * 1e9) - dp.SetSum(54.1) - dp.SetCount(98) - } else { - dp := metric.metric.Histogram().DataPoints().AppendEmpty() - dp.SetCount(0) - } - - return metric -} - -func metricPairToMetrics(mp []metricPair) pmetric.Metrics { - metrics := pmetric.NewMetrics() - metrics.ResourceMetrics().EnsureCapacity(len(mp)) - for num, record := range mp { - record.attributes.CopyTo(metrics.ResourceMetrics().AppendEmpty().Resource().Attributes()) - // TODO: Change metricPair to have an init metric func. - record.metric.CopyTo(metrics.ResourceMetrics().At(num).ScopeMetrics().AppendEmpty().Metrics().AppendEmpty()) - } - - return metrics -} - -func fieldsFromMap(s map[string]string) fields { - attrMap := pcommon.NewMap() - for k, v := range s { - attrMap.PutStr(k, v) - } - return newFields(attrMap) -} From 133ed3e7f8b95c46a97649825d0529d4cc2ef99e Mon Sep 17 00:00:00 2001 From: Dominik Rosiek Date: Tue, 9 Apr 2024 17:13:47 +0200 Subject: [PATCH 4/6] refactor!: move code from Sumo Logic repository Signed-off-by: Dominik Rosiek --- exporter/sumologicexporter/config.go | 206 ++- exporter/sumologicexporter/config_test.go | 105 +- exporter/sumologicexporter/exporter.go | 548 ++++--- exporter/sumologicexporter/exporter_test.go | 798 +++++++--- exporter/sumologicexporter/factory.go | 52 +- exporter/sumologicexporter/factory_test.go | 30 +- exporter/sumologicexporter/fields.go | 67 +- exporter/sumologicexporter/fields_test.go | 90 +- exporter/sumologicexporter/filter.go | 25 +- exporter/sumologicexporter/filter_test.go | 19 +- exporter/sumologicexporter/go.mod | 22 +- exporter/sumologicexporter/go.sum | 32 + .../sumologicexporter/prometheus_formatter.go | 248 ++- .../prometheus_formatter_test.go | 222 ++- exporter/sumologicexporter/sender.go | 801 +++++++--- exporter/sumologicexporter/sender_test.go | 1382 +++++++++++------ 16 files changed, 3137 insertions(+), 1510 deletions(-) diff --git a/exporter/sumologicexporter/config.go b/exporter/sumologicexporter/config.go index 16382c6942c7..aaae270b33e2 100644 --- a/exporter/sumologicexporter/config.go +++ b/exporter/sumologicexporter/config.go @@ -6,8 +6,12 @@ package sumologicexporter // import "github.com/open-telemetry/opentelemetry-col import ( "errors" "fmt" + "net/url" "time" + "go.opentelemetry.io/collector/component" + "go.opentelemetry.io/collector/config/configauth" + "go.opentelemetry.io/collector/config/configcompression" "go.opentelemetry.io/collector/config/confighttp" "go.opentelemetry.io/collector/config/configretry" "go.opentelemetry.io/collector/exporter/exporterhelper" @@ -21,7 +25,8 @@ type Config struct { // Compression encoding format, either empty string, gzip or deflate (default gzip) // Empty string means no compression - CompressEncoding CompressEncodingType `mapstructure:"compress_encoding"` + // NOTE: CompressEncoding is deprecated and will be removed in an upcoming release + CompressEncoding configcompression.Type `mapstructure:"compress_encoding"` // Max HTTP request body size in bytes before compression (if applied). // By default 1MB is recommended. MaxRequestBodySize int `mapstructure:"max_request_body_size"` @@ -30,41 +35,105 @@ type Config struct { // Format to post logs into Sumo. (default json) // * text - Logs will appear in Sumo Logic in text format. // * json - Logs will appear in Sumo Logic in json format. + // * otlp - Logs will be send in otlp format and will appear in Sumo Logic in text format. LogFormat LogFormatType `mapstructure:"log_format"` // Metrics related configuration - // The format of metrics you will be sending, either graphite or carbon2 or prometheus (Default is prometheus) - // Possible values are `carbon2` and `prometheus` + // The format of metrics you will be sending, either otlp or prometheus (Default is otlp) MetricFormat MetricFormatType `mapstructure:"metric_format"` - // Graphite template. - // Placeholders `%{attr_name}` will be replaced with attribute value for attr_name. - GraphiteTemplate string `mapstructure:"graphite_template"` - // List of regexes for attributes which should be send as metadata - MetadataAttributes []string `mapstructure:"metadata_attributes"` + // Decompose OTLP Histograms into individual metrics, similar to how they're represented in Prometheus format + DecomposeOtlpHistograms bool `mapstructure:"decompose_otlp_histograms"` + + // Traces related configuration + // The format of traces you will be sending, currently only otlp format is supported + TraceFormat TraceFormatType `mapstructure:"trace_format"` // Sumo specific options - // Desired source category. - // Useful if you want to override the source category configured for the source. - // Placeholders `%{attr_name}` will be replaced with attribute value for attr_name. - SourceCategory string `mapstructure:"source_category"` - // Desired source name. - // Useful if you want to override the source name configured for the source. - // Placeholders `%{attr_name}` will be replaced with attribute value for attr_name. - SourceName string `mapstructure:"source_name"` - // Desired host name. - // Useful if you want to override the source host configured for the source. - // Placeholders `%{attr_name}` will be replaced with attribute value for attr_name. - SourceHost string `mapstructure:"source_host"` // Name of the client Client string `mapstructure:"client"` + + // StickySessionEnabled defines if sticky session support is enable. + // By default this is false. + StickySessionEnabled bool `mapstructure:"sticky_session_enabled"` } -// createDefaultClientConfig returns default http client settings -func createDefaultClientConfig() confighttp.ClientConfig { - config := confighttp.NewDefaultClientConfig() - config.Timeout = defaultTimeout - return config +// CreateDefaultClientConfig returns default http client settings +func CreateDefaultClientConfig() confighttp.ClientConfig { + return confighttp.ClientConfig{ + Timeout: defaultTimeout, + Compression: DefaultCompressEncoding, + Auth: &configauth.Authentication{ + AuthenticatorID: component.NewID(Type), + }, + } +} + +func (cfg *Config) Validate() error { + + switch cfg.CompressEncoding { + case configcompression.TypeGzip: + case configcompression.TypeDeflate: + case NoCompression: + + default: + return fmt.Errorf("invalid compression encoding type: %v", cfg.ClientConfig.Compression) + } + + switch cfg.ClientConfig.Compression { + case configcompression.TypeGzip: + case configcompression.TypeDeflate: + case configcompression.TypeZstd: + case NoCompression: + + default: + return fmt.Errorf("invalid compression encoding type: %v", cfg.ClientConfig.Compression) + } + + if cfg.CompressEncoding != NoCompression && cfg.ClientConfig.Compression != DefaultCompressEncoding { + return fmt.Errorf("compress_encoding is deprecated and should not be used when compression is set to a non-default value") + } + + switch cfg.LogFormat { + case OTLPLogFormat: + case JSONFormat: + case TextFormat: + default: + return fmt.Errorf("unexpected log format: %s", cfg.LogFormat) + } + + switch cfg.MetricFormat { + case OTLPMetricFormat: + case PrometheusFormat: + case RemovedGraphiteFormat: + return fmt.Errorf("support for the graphite metric format was removed, please use prometheus or otlp instead") + case RemovedCarbon2Format: + return fmt.Errorf("support for the carbon2 metric format was removed, please use prometheus or otlp instead") + default: + return fmt.Errorf("unexpected metric format: %s", cfg.MetricFormat) + } + + switch cfg.TraceFormat { + case OTLPTraceFormat: + default: + return fmt.Errorf("unexpected trace format: %s", cfg.TraceFormat) + } + + if len(cfg.ClientConfig.Endpoint) == 0 && cfg.ClientConfig.Auth == nil { + return errors.New("no endpoint and no auth extension specified") + } + + if _, err := url.Parse(cfg.ClientConfig.Endpoint); err != nil { + return fmt.Errorf("failed parsing endpoint URL: %s; err: %w", + cfg.ClientConfig.Endpoint, err, + ) + } + + if err := cfg.QueueSettings.Validate(); err != nil { + return fmt.Errorf("queue settings has invalid configuration: %w", err) + } + + return nil } // LogFormatType represents log_format @@ -73,88 +142,55 @@ type LogFormatType string // MetricFormatType represents metric_format type MetricFormatType string +// TraceFormatType represents trace_format +type TraceFormatType string + // PipelineType represents type of the pipeline type PipelineType string -// CompressEncodingType represents type of the pipeline -type CompressEncodingType string - const ( // TextFormat represents log_format: text TextFormat LogFormatType = "text" // JSONFormat represents log_format: json JSONFormat LogFormatType = "json" - // GraphiteFormat represents metric_format: text - GraphiteFormat MetricFormatType = "graphite" - // Carbon2Format represents metric_format: json - Carbon2Format MetricFormatType = "carbon2" - // PrometheusFormat represents metric_format: json + // OTLPLogFormat represents log_format: otlp + OTLPLogFormat LogFormatType = "otlp" + // RemovedGraphiteFormat represents the no longer supported graphite metric format + RemovedGraphiteFormat MetricFormatType = "graphite" + // RemovedCarbon2Format represents the no longer supported carbon2 metric format + RemovedCarbon2Format MetricFormatType = "carbon2" + // PrometheusFormat represents metric_format: prometheus PrometheusFormat MetricFormatType = "prometheus" - // GZIPCompression represents compress_encoding: gzip - GZIPCompression CompressEncodingType = "gzip" - // DeflateCompression represents compress_encoding: deflate - DeflateCompression CompressEncodingType = "deflate" + // OTLPMetricFormat represents metric_format: otlp + OTLPMetricFormat MetricFormatType = "otlp" + // OTLPTraceFormat represents trace_format: otlp + OTLPTraceFormat TraceFormatType = "otlp" // NoCompression represents disabled compression - NoCompression CompressEncodingType = "" + NoCompression configcompression.Type = "" // MetricsPipeline represents metrics pipeline MetricsPipeline PipelineType = "metrics" // LogsPipeline represents metrics pipeline LogsPipeline PipelineType = "logs" + // TracesPipeline represents traces pipeline + TracesPipeline PipelineType = "traces" // defaultTimeout - defaultTimeout time.Duration = 5 * time.Second + defaultTimeout time.Duration = 30 * time.Second // DefaultCompress defines default Compress DefaultCompress bool = true // DefaultCompressEncoding defines default CompressEncoding - DefaultCompressEncoding CompressEncodingType = "gzip" + DefaultCompressEncoding configcompression.Type = "gzip" // DefaultMaxRequestBodySize defines default MaxRequestBodySize in bytes DefaultMaxRequestBodySize int = 1 * 1024 * 1024 // DefaultLogFormat defines default LogFormat - DefaultLogFormat LogFormatType = JSONFormat + DefaultLogFormat LogFormatType = OTLPLogFormat // DefaultMetricFormat defines default MetricFormat - DefaultMetricFormat MetricFormatType = PrometheusFormat - // DefaultSourceCategory defines default SourceCategory - DefaultSourceCategory string = "" - // DefaultSourceName defines default SourceName - DefaultSourceName string = "" - // DefaultSourceHost defines default SourceHost - DefaultSourceHost string = "" + DefaultMetricFormat MetricFormatType = OTLPMetricFormat // DefaultClient defines default Client DefaultClient string = "otelcol" - // DefaultGraphiteTemplate defines default template for Graphite - DefaultGraphiteTemplate string = "%{_metric_}" + // DefaultLogKey defines default LogKey value + DefaultLogKey string = "log" + // DefaultDropRoutingAttribute defines default DropRoutingAttribute + DefaultDropRoutingAttribute string = "" + // DefaultStickySessionEnabled defines default StickySessionEnabled value + DefaultStickySessionEnabled bool = false ) - -func (cfg *Config) Validate() error { - switch cfg.LogFormat { - case JSONFormat: - case TextFormat: - default: - return fmt.Errorf("unexpected log format: %s", cfg.LogFormat) - } - - switch cfg.MetricFormat { - case GraphiteFormat: - case Carbon2Format: - case PrometheusFormat: - default: - return fmt.Errorf("unexpected metric format: %s", cfg.MetricFormat) - } - - switch cfg.CompressEncoding { - case GZIPCompression: - case DeflateCompression: - case NoCompression: - default: - return fmt.Errorf("unexpected compression encoding: %s", cfg.CompressEncoding) - } - - if len(cfg.ClientConfig.Endpoint) == 0 { - return errors.New("endpoint is not set") - } - - if err := cfg.QueueSettings.Validate(); err != nil { - return fmt.Errorf("queue settings has invalid configuration: %w", err) - } - - return nil -} diff --git a/exporter/sumologicexporter/config_test.go b/exporter/sumologicexporter/config_test.go index 8eca2feb0951..c3d5493ff9c6 100644 --- a/exporter/sumologicexporter/config_test.go +++ b/exporter/sumologicexporter/config_test.go @@ -1,116 +1,113 @@ // Copyright The OpenTelemetry Authors // SPDX-License-Identifier: Apache-2.0 -package sumologicexporter +package sumologicexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/sumologicexporter" import ( + "errors" "testing" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" + "go.opentelemetry.io/collector/component" "go.opentelemetry.io/collector/config/confighttp" - "go.opentelemetry.io/collector/exporter/exporterhelper" ) -func TestConfigValidation(t *testing.T) { +func TestInitExporterInvalidConfiguration(t *testing.T) { testcases := []struct { - name string - cfg *Config - expectedErr string + name string + cfg *Config + expectedError error }{ { - name: "invalid log format", + name: "unexpected log format", + expectedError: errors.New("unexpected log format: test_format"), cfg: &Config{ - LogFormat: "test_format", - MetricFormat: "carbon2", - CompressEncoding: "gzip", + LogFormat: "test_format", + MetricFormat: "otlp", + TraceFormat: "otlp", ClientConfig: confighttp.ClientConfig{ Timeout: defaultTimeout, Endpoint: "test_endpoint", }, }, - expectedErr: "unexpected log format: test_format", }, { - name: "invalid metric format", + name: "unexpected metric format", + expectedError: errors.New("unexpected metric format: test_format"), cfg: &Config{ LogFormat: "json", MetricFormat: "test_format", ClientConfig: confighttp.ClientConfig{ - Timeout: defaultTimeout, - Endpoint: "test_endpoint", + Timeout: defaultTimeout, + Endpoint: "test_endpoint", + Compression: "gzip", }, - CompressEncoding: "gzip", }, - expectedErr: "unexpected metric format: test_format", }, { - name: "invalid compress encoding", + name: "unsupported Carbon2 metrics format", + expectedError: errors.New("support for the carbon2 metric format was removed, please use prometheus or otlp instead"), cfg: &Config{ - LogFormat: "json", - MetricFormat: "carbon2", - CompressEncoding: "test_format", + LogFormat: "json", + MetricFormat: "carbon2", ClientConfig: confighttp.ClientConfig{ - Timeout: defaultTimeout, - Endpoint: "test_endpoint", + Timeout: defaultTimeout, + Endpoint: "test_endpoint", + Compression: "gzip", }, }, - expectedErr: "unexpected compression encoding: test_format", }, { - name: "invalid endpoint", + name: "unsupported Graphite metrics format", + expectedError: errors.New("support for the graphite metric format was removed, please use prometheus or otlp instead"), cfg: &Config{ - LogFormat: "json", - MetricFormat: "carbon2", - CompressEncoding: "gzip", + LogFormat: "json", + MetricFormat: "graphite", ClientConfig: confighttp.ClientConfig{ - Timeout: defaultTimeout, + Timeout: defaultTimeout, + Endpoint: "test_endpoint", + Compression: "gzip", }, }, - expectedErr: "endpoint is not set", }, { - name: "invalid log format", + name: "unexpected trace format", + expectedError: errors.New("unexpected trace format: text"), cfg: &Config{ - LogFormat: "json", - MetricFormat: "carbon2", - CompressEncoding: "gzip", + LogFormat: "json", + MetricFormat: "otlp", + TraceFormat: "text", ClientConfig: confighttp.ClientConfig{ - Timeout: defaultTimeout, - Endpoint: "test_endpoint", - }, - QueueSettings: exporterhelper.QueueSettings{ - Enabled: true, - QueueSize: -10, + Timeout: defaultTimeout, + Endpoint: "test_endpoint", + Compression: "gzip", }, }, - expectedErr: "queue settings has invalid configuration: queue size must be positive", }, { - name: "valid config", + name: "no endpoint and no auth extension specified", + expectedError: errors.New("no endpoint and no auth extension specified"), cfg: &Config{ - LogFormat: "json", - MetricFormat: "carbon2", - CompressEncoding: "gzip", + LogFormat: "json", + MetricFormat: "otlp", + TraceFormat: "otlp", ClientConfig: confighttp.ClientConfig{ - Timeout: defaultTimeout, - Endpoint: "test_endpoint", + Timeout: defaultTimeout, + Compression: "gzip", }, }, - expectedErr: "", }, } for _, tc := range testcases { + tc := tc t.Run(tc.name, func(t *testing.T) { + err := component.ValidateConfig(tc.cfg) - err := tc.cfg.Validate() - - if tc.expectedErr == "" { - assert.NoError(t, err) + if tc.expectedError != nil { + assert.EqualError(t, err, tc.expectedError.Error()) } else { - require.Error(t, err) - assert.EqualError(t, err, tc.expectedErr) + assert.NoError(t, err) } }) } diff --git a/exporter/sumologicexporter/exporter.go b/exporter/sumologicexporter/exporter.go index 64347cc22313..546b5bb8fc64 100644 --- a/exporter/sumologicexporter/exporter.go +++ b/exporter/sumologicexporter/exporter.go @@ -8,6 +8,10 @@ import ( "errors" "fmt" "net/http" + "net/url" + "path" + "strings" + "sync" "go.opentelemetry.io/collector/component" "go.opentelemetry.io/collector/consumer/consumererror" @@ -16,79 +20,83 @@ import ( "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" + "go.opentelemetry.io/collector/pdata/ptrace" + "go.uber.org/zap" + + "github.com/open-telemetry/opentelemetry-collector-contrib/extension/sumologicextension" ) -type sumologicexporter struct { - sources sourceFormats - config *Config - client *http.Client - filter filter - prometheusFormatter prometheusFormatter - graphiteFormatter graphiteFormatter - settings component.TelemetrySettings -} +const ( + logsDataUrl = "/api/v1/collector/logs" + metricsDataUrl = "/api/v1/collector/metrics" + tracesDataUrl = "/api/v1/collector/traces" +) -func initExporter(cfg *Config, settings component.TelemetrySettings) (*sumologicexporter, error) { +type sumologicexporter struct { + config *Config + host component.Host + logger *zap.Logger - if cfg.MetricFormat == GraphiteFormat { - settings.Logger.Warn("`metric_format: graphite` nad `graphite_template` are deprecated and are going to be removed in the future. See https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/exporter/sumologicexporter#migration-to-new-architecture for more information") - } + clientLock sync.RWMutex + client *http.Client - if cfg.MetricFormat == Carbon2Format { - settings.Logger.Warn("`metric_format: carbon` is deprecated and is going to be removed in the future. See https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/exporter/sumologicexporter#migration-to-new-architecture for more information") - } + prometheusFormatter prometheusFormatter - if len(cfg.MetadataAttributes) > 0 { - settings.Logger.Warn("`metadata_attributes: []` is deprecated and is going to be removed in the future. See https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/exporter/sumologicexporter#migration-to-new-architecture for more information") - } + // Lock around data URLs is needed because the reconfiguration of the exporter + // can happen asynchronously whenever the exporter is re registering. + dataUrlsLock sync.RWMutex + dataUrlMetrics string + dataUrlLogs string + dataUrlTraces string - if cfg.SourceCategory != "" { - settings.Logger.Warn("`source_category: