forked from grafana/agent
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
otelcol.exporter.otlphttp: new component (grafana#2429)
Signed-off-by: Paschalis Tsilias <paschalis.tsilias@grafana.com>
- Loading branch information
1 parent
6436b35
commit 2f178bb
Showing
11 changed files
with
549 additions
and
67 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| // Package otlphttp provides an otelcol.exporter.otlphttp component. | ||
| package otlphttp | ||
|
|
||
| import ( | ||
| "errors" | ||
| "time" | ||
|
|
||
| "github.com/grafana/agent/component" | ||
| "github.com/grafana/agent/component/otelcol" | ||
| "github.com/grafana/agent/component/otelcol/exporter" | ||
| "github.com/grafana/agent/pkg/river" | ||
| otelcomponent "go.opentelemetry.io/collector/component" | ||
| otelconfig "go.opentelemetry.io/collector/config" | ||
| "go.opentelemetry.io/collector/exporter/otlphttpexporter" | ||
| ) | ||
|
|
||
| func init() { | ||
| component.Register(component.Registration{ | ||
| Name: "otelcol.exporter.otlphttp", | ||
| Args: Arguments{}, | ||
| Exports: otelcol.ConsumerExports{}, | ||
|
|
||
| Build: func(opts component.Options, args component.Arguments) (component.Component, error) { | ||
| fact := otlphttpexporter.NewFactory() | ||
| return exporter.New(opts, fact, args.(Arguments)) | ||
| }, | ||
| }) | ||
| } | ||
|
|
||
| // Arguments configures the otelcol.exporter.otlphttp component. | ||
| type Arguments struct { | ||
| Client HTTPClientArguments `river:"client,block"` | ||
| Queue otelcol.QueueArguments `river:"sending_queue,block,optional"` | ||
| Retry otelcol.RetryArguments `river:"retry_on_failure,block,optional"` | ||
|
|
||
| // The URLs to send metrics/logs/traces to. If omitted the exporter will | ||
| // use Client.Endpoint by appending "/v1/metrics", "/v1/logs" or | ||
| // "/v1/traces", respectively. If set, these settings override | ||
| // Client.Endpoint for the corresponding signal. | ||
| TracesEndpoint string `river:"traces_endpoint,attr,optional"` | ||
| MetricsEndpoint string `river:"metrics_endpoint,attr,optional"` | ||
| LogsEndpoint string `river:"logs_endpoint,attr,optional"` | ||
| } | ||
|
|
||
| var ( | ||
| _ river.Unmarshaler = (*Arguments)(nil) | ||
| _ river.Unmarshaler = (*HTTPClientArguments)(nil) | ||
| _ exporter.Arguments = Arguments{} | ||
| ) | ||
|
|
||
| // DefaultArguments holds default values for Arguments. | ||
| var DefaultArguments = Arguments{ | ||
| Queue: otelcol.DefaultQueueArguments, | ||
| Retry: otelcol.DefaultRetryArguments, | ||
| Client: DefaultHTTPClientArguments, | ||
| } | ||
|
|
||
| // UnmarshalRiver implements river.Unmarshaler. | ||
| func (args *Arguments) UnmarshalRiver(f func(interface{}) error) error { | ||
| *args = DefaultArguments | ||
| type arguments Arguments | ||
| err := f((*arguments)(args)) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return args.Validate() | ||
| } | ||
|
|
||
| // Convert implements exporter.Arguments. | ||
| func (args Arguments) Convert() otelconfig.Exporter { | ||
| return &otlphttpexporter.Config{ | ||
| ExporterSettings: otelconfig.NewExporterSettings(otelconfig.NewComponentID("otlp")), | ||
| HTTPClientSettings: *(*otelcol.HTTPClientArguments)(&args.Client).Convert(), | ||
| QueueSettings: *args.Queue.Convert(), | ||
| RetrySettings: *args.Retry.Convert(), | ||
| } | ||
| } | ||
|
|
||
| // Extensions implements exporter.Arguments. | ||
| func (args Arguments) Extensions() map[otelconfig.ComponentID]otelcomponent.Extension { | ||
| return (*otelcol.HTTPClientArguments)(&args.Client).Extensions() | ||
| } | ||
|
|
||
| // Exporters implements exporter.Arguments. | ||
| func (args Arguments) Exporters() map[otelconfig.DataType]map[otelconfig.ComponentID]otelcomponent.Exporter { | ||
| return nil | ||
| } | ||
|
|
||
| // Validate returns an error if the configuration is invalid. | ||
| func (args *Arguments) Validate() error { | ||
| if args.Client.Endpoint == "" && args.TracesEndpoint == "" && args.MetricsEndpoint == "" && args.LogsEndpoint == "" { | ||
| return errors.New("at least one endpoint must be specified") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // HTTPClientArguments is used to configure otelcol.exporter.otlphttp with | ||
| // component-specific defaults. | ||
| type HTTPClientArguments otelcol.HTTPClientArguments | ||
|
|
||
| // Default server settings. | ||
| var ( | ||
| DefaultMaxIddleConns = 100 | ||
| DefaultIdleConnTimeout = 90 * time.Second | ||
| DefaultHTTPClientArguments = HTTPClientArguments{ | ||
| MaxIdleConns: &DefaultMaxIddleConns, | ||
| IdleConnTimeout: &DefaultIdleConnTimeout, | ||
|
|
||
| Timeout: 30 * time.Second, | ||
| Headers: map[string]string{}, | ||
| Compression: otelcol.CompressionTypeGzip, | ||
| ReadBufferSize: 0, | ||
| WriteBufferSize: 512 * 1024, | ||
| } | ||
| ) | ||
|
|
||
| // UnmarshalRiver implements river.Unmarshaler and supplies defaults. | ||
| func (args *HTTPClientArguments) UnmarshalRiver(f func(interface{}) error) error { | ||
| *args = DefaultHTTPClientArguments | ||
| type arguments HTTPClientArguments | ||
| return f((*arguments)(args)) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| package otlphttp_test | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "io/ioutil" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log/level" | ||
| "github.com/grafana/agent/component/otelcol" | ||
| "github.com/grafana/agent/component/otelcol/exporter/otlphttp" | ||
| "github.com/grafana/agent/pkg/flow/componenttest" | ||
| "github.com/grafana/agent/pkg/river" | ||
| "github.com/grafana/agent/pkg/util" | ||
| "github.com/grafana/dskit/backoff" | ||
| "github.com/stretchr/testify/require" | ||
| "go.opentelemetry.io/collector/pdata/ptrace" | ||
| ) | ||
|
|
||
| // Test performs a basic integration test which runs the | ||
| // otelcol.exporter.otlphttp component and ensures that it can pass data to an | ||
| // OTLP HTTP server. | ||
| func Test(t *testing.T) { | ||
| ch := make(chan ptrace.Traces) | ||
| srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| b, _ := ioutil.ReadAll(r.Body) | ||
| trace, _ := ptrace.NewProtoUnmarshaler().UnmarshalTraces(b) | ||
| require.Equal(t, 1, trace.SpanCount()) | ||
| name := trace.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Name() | ||
| require.Equal(t, "TestSpan", name) | ||
| ch <- trace | ||
| w.WriteHeader(http.StatusOK) | ||
| })) | ||
| defer srv.Close() | ||
|
|
||
| ctx := componenttest.TestContext(t) | ||
| l := util.TestLogger(t) | ||
|
|
||
| ctrl, err := componenttest.NewControllerFromID(l, "otelcol.exporter.otlphttp") | ||
| require.NoError(t, err) | ||
|
|
||
| cfg := fmt.Sprintf(` | ||
| client { | ||
| endpoint = "%s" | ||
| compression = "none" | ||
| tls { | ||
| insecure = true | ||
| insecure_skip_verify = true | ||
| } | ||
| } | ||
| `, srv.URL) | ||
| var args otlphttp.Arguments | ||
| require.NoError(t, river.Unmarshal([]byte(cfg), &args)) | ||
|
|
||
| go func() { | ||
| err := ctrl.Run(ctx, args) | ||
| require.NoError(t, err) | ||
| }() | ||
|
|
||
| require.NoError(t, ctrl.WaitRunning(time.Second), "component never started") | ||
| require.NoError(t, ctrl.WaitExports(time.Second), "component never exported anything") | ||
|
|
||
| // Send traces in the background to our exporter. | ||
| go func() { | ||
| exports := ctrl.Exports().(otelcol.ConsumerExports) | ||
|
|
||
| bo := backoff.New(ctx, backoff.Config{ | ||
| MinBackoff: 10 * time.Millisecond, | ||
| MaxBackoff: 100 * time.Millisecond, | ||
| }) | ||
| for bo.Ongoing() { | ||
| err := exports.Input.ConsumeTraces(ctx, createTestTraces()) | ||
| if err != nil { | ||
| level.Error(l).Log("msg", "failed to send traces", "err", err) | ||
| bo.Wait() | ||
| continue | ||
| } | ||
|
|
||
| return | ||
| } | ||
| }() | ||
|
|
||
| // Wait for our exporter to finish and pass data to our HTTP server. | ||
| select { | ||
| case <-time.After(time.Second): | ||
| require.FailNow(t, "failed waiting for traces") | ||
| case tr := <-ch: | ||
| require.Equal(t, 1, tr.SpanCount()) | ||
| } | ||
| } | ||
|
|
||
| func createTestTraces() ptrace.Traces { | ||
| // Matches format from the protobuf definition: | ||
| // https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto | ||
| var bb = `{ | ||
| "resource_spans": [{ | ||
| "scope_spans": [{ | ||
| "spans": [{ | ||
| "name": "TestSpan" | ||
| }] | ||
| }] | ||
| }] | ||
| }` | ||
|
|
||
| data, err := ptrace.NewJSONUnmarshaler().UnmarshalTraces([]byte(bb)) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return data | ||
| } |
Oops, something went wrong.