mirror of
https://github.com/hanzoai/otel-collector.git
synced 2026-11-03 02:20:04 +00:00
Raj Kamal SinghandGitHub
afd3bbf56e
Brings in a copy of collector simulator from signoz query service. In a follow up, query service collector simulator will be removed and query service will starting using this collector simulator This change will unblock work on issues like https://github.com/SigNoz/engineering-pod/issues/1112 This change will also make it easier to maintain collector simulator whenever signoz-otel-collector is upgraded to a higher upstream version (eg: https://github.com/SigNoz/signoz/pull/6257/files#diff-943188c9097efb17b1b87d9f7166345f1520bb7277eda120832fc1897bf4c490R144) The code has been copied over as-is from https://github.com/SigNoz/signoz/tree/develop/pkg/query-service/collectorsimulator with minor changes to use `error` instead of `*model.ApiError`
65 lines
1.6 KiB
Go
65 lines
1.6 KiB
Go
package inmemoryreceiver
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"go.opentelemetry.io/collector/component"
|
|
"go.opentelemetry.io/collector/consumer"
|
|
"go.opentelemetry.io/collector/pdata/plog"
|
|
)
|
|
|
|
// In memory receiver for testing and simulation
|
|
type InMemoryReceiver struct {
|
|
// Unique identifier for the receiver.
|
|
id string
|
|
|
|
nextConsumer consumer.Logs
|
|
}
|
|
|
|
func (r *InMemoryReceiver) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
|
|
return r.nextConsumer.ConsumeLogs(ctx, ld)
|
|
}
|
|
|
|
func (r *InMemoryReceiver) Capabilities() consumer.Capabilities {
|
|
return consumer.Capabilities{MutatesData: false}
|
|
}
|
|
|
|
// Keep track of all receiver instances in the process.
|
|
// Useful for getting a hold of the receiver in scenarios where one doesn't
|
|
// create the instances. Eg: bringing up a collector service from collector config
|
|
var allReceiverInstances map[string]*InMemoryReceiver
|
|
var allReceiversLock sync.Mutex
|
|
|
|
func init() {
|
|
allReceiverInstances = make(map[string]*InMemoryReceiver)
|
|
}
|
|
|
|
func CleanupInstance(receiverId string) {
|
|
allReceiversLock.Lock()
|
|
defer allReceiversLock.Unlock()
|
|
delete(allReceiverInstances, receiverId)
|
|
}
|
|
|
|
func (r *InMemoryReceiver) Start(ctx context.Context, host component.Host) error {
|
|
allReceiversLock.Lock()
|
|
defer allReceiversLock.Unlock()
|
|
|
|
if allReceiverInstances[r.id] != nil {
|
|
return fmt.Errorf("receiver with id %s is already running", r.id)
|
|
}
|
|
|
|
allReceiverInstances[r.id] = r
|
|
return nil
|
|
}
|
|
|
|
func (r *InMemoryReceiver) Shutdown(ctx context.Context) error {
|
|
CleanupInstance(r.id)
|
|
return nil
|
|
}
|
|
|
|
func GetReceiverInstance(id string) *InMemoryReceiver {
|
|
return allReceiverInstances[id]
|
|
}
|