[signalfx_correlation] Add signalfx_correlation exporter skeleton (#1332)

* [signalfx_correlation] Add signalfx_correlation exporter skeleton

This is for moving the correlation out of sapmexporter into a dedicated
exporter so that the correlation can be used even when sapm isn't (for example,
on an agent that is exporting in otlp to a gateway instead of sapm.)

* fix readme

Author: jrcamp

exporter/signalfxcorrelationexporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

exporter/signalfxcorrelationexporter/README.md

@@ -0,0 +1,38 @@
+# SignalFx Correlation Exporter
+
+This exporter observes traces being sent for their service and environment. When a new service or environment is
+seen it associates the source (e.g. host or pod) to that source or environment in SignalFx metrics so that those
+metrics can be filtered by the service and environment.
+
+Supported pipeline types: traces
+
+## Configuration
+
+The following configuration options are required:
+
+- `endpoint` (required, no default): This is the base URL for API requests (e.g. https://api.signalfx.com).
+- `access_token` (no default): The authentication token provided by SignalFx.
+- `timeout` (default = 5s): Is the timeout for every attempt to send data to the backend.
+- `stale_service_timeout` (default = 5 minutes): How long to wait after a span's service name is last seen before uncorrelating it.
+- `max_requests` (default = 20): Max HTTP requests to be made in parallel.
+- `max_buffered` (default = 10,000): Max number of correlation updates that can be buffered before updates are dropped.
+- `max_retries` (default = 2): Max number of retries that will be made for failed correlation updates.
+- `log_updates` (default = false): Whether or not to log correlation updates to dimensions (at `DEBUG` level).
+- `retry_delay` (default = 30 seconds): How long to wait between retries.
+- `cleanup_interval` (default = 1 minute): How frequently to purge duplicate requests.
+- `sync_attributes` (default = `{"k8s.pod.uid": "k8s.pod.uid", "container.id": "container.id"}`) Map containing key of the attribute to read from spans to sync to dimensions specified as the value.
+
+Example:
+
+```yaml
+exporters:
+  signalfx_correlation:
+    access_token: YOUR_ACCESS_TOKEN
+    endpoint: https://api.YOUR_SIGNALFX_REALM.signalfx.com
+```
+
+The full list of settings exposed for this exporter are documented [here](config.go)
+with detailed sample configurations [here](testdata/config.yaml).
+
+This exporter also offers proxy support as documented
+[here](https://github.com/open-telemetry/opentelemetry-collector/tree/master/exporter#proxy-support).

exporter/signalfxcorrelationexporter/config.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package signalfxcorrelationexporter
+
+import (
+	"errors"
+	"net/url"
+	"time"
+
+	"github.com/signalfx/signalfx-agent/pkg/apm/correlations"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/config/configmodels"
+)
+
+// Config defines configuration for signalfx_correlation exporter.
+type Config struct {
+	configmodels.ExporterSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	confighttp.HTTPClientSettings `mapstructure:",squash"`
+	correlations.Config           `mapstructure:",squash"`
+
+	// How long to wait after a trace span's service name is last seen before
+	// uncorrelating that service.
+	StaleServiceTimeout time.Duration `mapstructure:"stale_service_timeout"`
+	// SyncAttributes is a key of the span attribute name to sync to the dimension as the value.
+	SyncAttributes map[string]string `mapstructure:"sync_attributes"`
+
+	// AccessToken is the authentication token provided by SignalFx.
+	AccessToken string `mapstructure:"access_token"`
+}
+
+func (c *Config) validate() error {
+	if c.Endpoint == "" {
+		return errors.New("`endpoint` not specified")
+	}
+
+	_, err := url.Parse(c.Endpoint)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

exporter/signalfxcorrelationexporter/config_test.go

@@ -0,0 +1,84 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package signalfxcorrelationexporter
+
+import (
+	"path"
+	"testing"
+	"time"
+
+	"github.com/signalfx/signalfx-agent/pkg/apm/correlations"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	facotries, err := componenttest.ExampleComponents()
+	assert.Nil(t, err)
+
+	factory := NewFactory()
+	facotries.Exporters[configmodels.Type(typeStr)] = factory
+	cfg, err := configtest.LoadConfigFile(
+		t, path.Join(".", "testdata", "config.yaml"), facotries,
+	)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	assert.Equal(t, len(cfg.Exporters), 2)
+
+	r0 := cfg.Exporters["signalfx_correlation"]
+	assert.Equal(t, r0, factory.CreateDefaultConfig())
+
+	r1 := cfg.Exporters["signalfx_correlation/configured"].(*Config)
+	assert.Equal(t, r1,
+		&Config{
+			ExporterSettings:    configmodels.ExporterSettings{TypeVal: configmodels.Type(typeStr), NameVal: "signalfx_correlation/configured"},
+			HTTPClientSettings:  confighttp.HTTPClientSettings{Endpoint: "https://api.signalfx.com", Timeout: 10 * time.Second},
+			AccessToken:         "abcd1234",
+			StaleServiceTimeout: 5 * time.Minute,
+			SyncAttributes: map[string]string{
+				"k8s.pod.uid":  "k8s.pod.uid",
+				"container.id": "container.id",
+			},
+			Config: correlations.Config{
+				MaxRequests:     20,
+				MaxBuffered:     10_000,
+				MaxRetries:      2,
+				LogUpdates:      false,
+				RetryDelay:      30 * time.Second,
+				CleanupInterval: 1 * time.Minute,
+			},
+		})
+}
+
+func TestInvalidConfig(t *testing.T) {
+	invalid := Config{
+		AccessToken: "abcd1234",
+	}
+	noEndpointErr := invalid.validate()
+	require.Error(t, noEndpointErr)
+
+	invalid = Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{Endpoint: ":123:456"},
+		AccessToken:        "abcd1234",
+	}
+	invalidURLErr := invalid.validate()
+	require.Error(t, invalidURLErr)
+}

exporter/signalfxcorrelationexporter/doc.go

@@ -0,0 +1,16 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package signalfxcorrelationexporter performs span to metric correlation for SignalFx.
+package signalfxcorrelationexporter

exporter/signalfxcorrelationexporter/exporter.go

@@ -0,0 +1,63 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package signalfxcorrelationexporter
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.uber.org/zap"
+)
+
+// corrExporter is a wrapper struct of the correlation exporter
+type corrExporter struct {
+	logger *zap.Logger
+	config *Config
+}
+
+func (se *corrExporter) Shutdown(context.Context) error {
+	return nil
+}
+
+func newCorrExporter(cfg *Config, params component.ExporterCreateParams) (corrExporter, error) {
+	err := cfg.validate()
+	if err != nil {
+		return corrExporter{}, err
+	}
+
+	return corrExporter{
+		logger: params.Logger,
+		config: cfg,
+	}, err
+}
+
+func newTraceExporter(cfg *Config, params component.ExporterCreateParams) (component.TraceExporter, error) {
+	se, err := newCorrExporter(cfg, params)
+	if err != nil {
+		return nil, err
+	}
+
+	return exporterhelper.NewTraceExporter(
+		cfg,
+		se.pushTraceData,
+		exporterhelper.WithShutdown(se.Shutdown))
+}
+
+// pushTraceData processes traces to extract services and environments to associate them to their emitting host/pods.
+func (se *corrExporter) pushTraceData(ctx context.Context, td pdata.Traces) (droppedSpansCount int, err error) {
+	return 0, nil
+}

exporter/signalfxcorrelationexporter/exporter_test.go

@@ -0,0 +1,67 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package signalfxcorrelationexporter
+
+import (
+	"context"
+	"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/config/configmodels"
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.uber.org/zap"
+)
+
+func TestCreateTraceExporter(t *testing.T) {
+	config := &Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{Endpoint: "http://localhost"},
+		ExporterSettings:   configmodels.ExporterSettings{TypeVal: configmodels.Type(typeStr), NameVal: "signalfx_correlation/configured"},
+		AccessToken:        "abcd1234",
+	}
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
+
+	te, err := newCorrExporter(config, params)
+	assert.Nil(t, err)
+	assert.NotNil(t, te, "failed to create trace exporter")
+
+	assert.NoError(t, te.Shutdown(context.Background()), "trace exporter shutdown failed")
+}
+
+func TestCreateTraceExporterWithInvalidConfig(t *testing.T) {
+	config := &Config{}
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
+	te, err := newTraceExporter(config, params)
+	require.Error(t, err)
+	assert.Nil(t, te)
+}
+
+func TestExporterConsumeTraces(t *testing.T) {
+	config := &Config{
+		HTTPClientSettings: confighttp.HTTPClientSettings{Endpoint: "http://localhost"},
+		ExporterSettings:   configmodels.ExporterSettings{TypeVal: configmodels.Type(typeStr), NameVal: "signalfx_correlation/configured"},
+		AccessToken:        "abcd1234",
+	}
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
+
+	te, err := newTraceExporter(config, params)
+	assert.Nil(t, err)
+	assert.NotNil(t, te, "failed to create trace exporter")
+	defer te.Shutdown(context.Background())
+
+	assert.NoError(t, te.ConsumeTraces(context.Background(), pdata.NewTraces()))
+}