-
Notifications
You must be signed in to change notification settings - Fork 3.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
obsservice: migrate grpc EventIngester to fit new architecture
In the original design of the obsservice, exported events were intended to be written directly to storage. The idea was that exported events would experience minimal transformation once ingested, meaning that work done to "package" events properly was left up to the exporting client (CRDB). The obsservice would then store the ingested invents into a target storage. This concept of target storage has been removed for now as part of this patch. In the new architecture, exported events are more "raw", and we expect the obsservice to heavily transform & aggregate the data externally, where the aggregated results are flushed to storage instead. This patch takes the pre-existing gRPC events ingester, and modifies it to meet the new architecture. The events ingester will now be provided with a consumer with which it can feed ingested events into the broader pipeline. It is no longer the responsibility of the ingester to write ingested events to storage. For now, we use a simple STDOUT consumer that writes all ingested events to STDOUT, but in the future, this will be a more legitimate component - part of a chain that eventually buffers ingested events for aggregation. Release note: none
- Loading branch information
1 parent
3fb55d0
commit 946d774
Showing
21 changed files
with
801 additions
and
453 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
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,81 @@ | ||
| // Copyright 2022 The Cockroach 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 | ||
|
|
||
| package ingest | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/cockroachdb/cockroach/pkg/obsservice/obslib" | ||
| "github.com/cockroachdb/cockroach/pkg/obsservice/obslib/transform" | ||
| logspb "github.com/cockroachdb/cockroach/pkg/obsservice/obspb/opentelemetry-proto/collector/logs/v1" | ||
| "github.com/cockroachdb/cockroach/pkg/util/log" | ||
| "github.com/cockroachdb/cockroach/pkg/util/timeutil" | ||
| "github.com/cockroachdb/errors" | ||
| ) | ||
|
|
||
| // EventIngester implements the OTLP Logs gRPC service, accepting connections | ||
| // and ingesting events. | ||
| type EventIngester struct { | ||
| consumer obslib.EventConsumer | ||
| timeSource timeutil.TimeSource | ||
| } | ||
|
|
||
| var _ logspb.LogsServiceServer = &EventIngester{} | ||
|
|
||
| // MakeEventIngester creates a new EventIngester. Callers can optionally | ||
| // provide a timeutil.TimeSource which will be used when determining ingestion | ||
| // timestamps. If nil is provided, timeutil.DefaultTimeSource will be used. | ||
| func MakeEventIngester( | ||
| _ context.Context, consumer obslib.EventConsumer, timeSource timeutil.TimeSource, | ||
| ) *EventIngester { | ||
| if timeSource == nil { | ||
| timeSource = timeutil.DefaultTimeSource{} | ||
| } | ||
| return &EventIngester{ | ||
| consumer: consumer, | ||
| timeSource: timeSource, | ||
| } | ||
| } | ||
|
|
||
| // Export implements the LogsServiceServer gRPC service. | ||
| // | ||
| // NB: "Export" is a bit of a misnomer here. On the client side, | ||
| // it makes sense, but on this end of the wire, we are *Ingesting* | ||
| // events, not exporting them. This is simply the receiving end | ||
| // of that process. This is done to maintain compatibility between | ||
| // the OpenTelemetry Collector and our EventsExporter. | ||
| func (e *EventIngester) Export( | ||
| ctx context.Context, request *logspb.ExportLogsServiceRequest, | ||
| ) (*logspb.ExportLogsServiceResponse, error) { | ||
| if err := e.unpackAndConsumeEvents(ctx, request); err != nil { | ||
| log.Errorf(ctx, "error consuming events: %v", err) | ||
| return nil, err | ||
| } | ||
| return &logspb.ExportLogsServiceResponse{}, nil | ||
| } | ||
|
|
||
| // TODO(abarganier): Add context cancellation here to cap transformation/unpack time. | ||
| // TODO(abarganier): Add metric to track context cancellations (counter tracking failed transformations) | ||
| func (e *EventIngester) unpackAndConsumeEvents( | ||
| ctx context.Context, request *logspb.ExportLogsServiceRequest, | ||
| ) error { | ||
| ingestTime := e.timeSource.Now() | ||
| var retErr error = nil | ||
| for _, resource := range request.ResourceLogs { | ||
| for _, scopeLogs := range resource.ScopeLogs { | ||
| for _, logRecord := range scopeLogs.LogRecords { | ||
| transformed := transform.LogRecordToEvent(ingestTime, resource.Resource, scopeLogs.Scope, logRecord) | ||
| if err := e.consumer.Consume(ctx, transformed); err != nil { | ||
| retErr = errors.CombineErrors(retErr, err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return retErr | ||
| } |
Oops, something went wrong.