sinker

Documentation

Index

• Constants
• Variables
• type Service
  • func New(logger *zap.Logger, pubSub mfnats.PubSub, esclient *redis.Client, ...) Service
• type SinkerService
  • func (svc SinkerService) ExtractAgent(ctx context.Context, channelID string) (*pb.AgentInfoRes, error)
  • func (svc SinkerService) GetSinks(agent fleet.Agent, agentMetricsRPCPayload fleet.AgentMetricsRPCPayload, ...) error
  • func (svc SinkerService) SinkPolicy(agent fleet.Agent, metricsPayload fleet.AgentMetricsRPCPayload, ...)
  • func (svc SinkerService) Start() error
  • func (svc SinkerService) Stop() error

Constants

const (
	CheckerFreq    = 5 * time.Minute
	DefaultTimeout = 30 * time.Minute
)

const (
	BackendMetricsTopic = "be.*.m.>"
	OtelMetricsTopic    = "otlp.*.m.>"
	MaxMsgPayloadSize   = 1048 * 1000
)

Variables

var (
	ErrPayloadTooBig = errors.New("payload too big")
	ErrNotFound      = errors.New("non-existent entity")
)

Types

type Service

type Service interface {
	// Start set up communication with the message bus to communicate with agents
	Start() error
	// Stop end communication with the message bus
	Stop() error
}

func New

func New(logger *zap.Logger,
	pubSub mfnats.PubSub,
	esclient *redis.Client,
	configRepo config.ConfigRepo,
	policiesClient policiespb.PolicyServiceClient,
	fleetClient fleetpb.FleetServiceClient,
	sinksClient sinkspb.SinkServiceClient,
	otelKafkaUrl string,
	enableOtel bool,
	requestGauge metrics.Gauge,
	requestCounter metrics.Counter,
	inputCounter metrics.Counter,
) Service

New instantiates the sinker service implementation.

type SinkerService

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

func (SinkerService) ExtractAgent

func (svc SinkerService) ExtractAgent(ctx context.Context, channelID string) (*pb.AgentInfoRes, error)

func (SinkerService) GetSinks

func (svc SinkerService) GetSinks(agent fleet.Agent, agentMetricsRPCPayload fleet.AgentMetricsRPCPayload, datasetSinkIDs map[string]bool) error

func (SinkerService) SinkPolicy

func (svc SinkerService) SinkPolicy(agent fleet.Agent, metricsPayload fleet.AgentMetricsRPCPayload, datasetSinkIDs map[string]bool, tsList []prometheus.TimeSeries)

func (SinkerService) Start

func (svc SinkerService) Start() error

func (SinkerService) Stop

func (svc SinkerService) Stop() error