bridgeservice

Documentation

Index

• type BridgeService
• type SinkerOtelBridgeService
  • func NewBridgeService(logger *zap.Logger, defaultCacheExpiration time.Duration, ...) SinkerOtelBridgeService
  • func (bs *SinkerOtelBridgeService) ExtractAgent(ctx context.Context, channelID string) (*fleetpb.AgentInfoRes, error)
  • func (bs *SinkerOtelBridgeService) GetPolicyName(ctx context.Context, policyId, ownerID string) (*policiespb.PolicyRes, error)
  • func (bs *SinkerOtelBridgeService) GetSinkIdsFromDatasetIDs(ctx context.Context, mfOwnerId string, datasetIDs []string) (map[string]string, error)
  • func (bs *SinkerOtelBridgeService) IncrementMessageCounter(publisher, subtopic, channel, protocol string)
  • func (bs *SinkerOtelBridgeService) NotifyActiveSink(ctx context.Context, mfOwnerId, sinkId, size string) error

Types

type BridgeService

type BridgeService interface {
	ExtractAgent(ctx context.Context, channelID string) (*fleetpb.AgentInfoRes, error)
	GetPolicyName(ctx context.Context, policyId, ownerId string) (*policiespb.PolicyRes, error)
	GetDataSetsFromAgentGroups(ctx context.Context, mfOwnerId string, agentGroupIds []string) (map[string]string, error)
	NotifyActiveSink(ctx context.Context, mfOwnerId, sinkId, state, message string) error
	GetSinkIdsFromPolicyID(ctx context.Context, mfOwnerId string, policyID string) (map[string]string, error)
	IncreamentMessageCounter(publisher, subtopic, channel, protocol string)
}

type SinkerOtelBridgeService

type SinkerOtelBridgeService struct {
	// contains filtered or unexported fields
}

func NewBridgeService

func NewBridgeService(logger *zap.Logger,
	defaultCacheExpiration time.Duration,
	sinkActivity producer.SinkActivityProducer,
	policiesClient policiespb.PolicyServiceClient,
	sinksClient sinkspb.SinkServiceClient,
	fleetClient fleetpb.FleetServiceClient, messageInputCounter metrics.Counter) SinkerOtelBridgeService

func (*SinkerOtelBridgeService) ExtractAgent

func (bs *SinkerOtelBridgeService) ExtractAgent(ctx context.Context, channelID string) (*fleetpb.AgentInfoRes, error)

ExtractAgent retrieve agent info from fleet, or cache

func (*SinkerOtelBridgeService) GetPolicyName

func (bs *SinkerOtelBridgeService) GetPolicyName(ctx context.Context, policyId, ownerID string) (*policiespb.PolicyRes, error)

GetPolicyName retrieve policy info from policies service, or cache.

func (*SinkerOtelBridgeService) GetSinkIdsFromDatasetIDs

func (bs *SinkerOtelBridgeService) GetSinkIdsFromDatasetIDs(ctx context.Context, mfOwnerId string, datasetIDs []string) (map[string]string, error)

GetSinkIdsFromDatasetIDs retrieve sink_ids from datasets from policies service, or cache

func (*SinkerOtelBridgeService) IncrementMessageCounter

func (bs *SinkerOtelBridgeService) IncrementMessageCounter(publisher, subtopic, channel, protocol string)

IncrementMessageCounter add to our metrics the number of messages received

func (*SinkerOtelBridgeService) NotifyActiveSink

func (bs *SinkerOtelBridgeService) NotifyActiveSink(ctx context.Context, mfOwnerId, sinkId, size string) error

NotifyActiveSink notify the sinker that a sink is active