opamp

Documentation

Overview

Package opamp provides interfaces for managing agents' OpAMP connections

Index

• Constants
• Variables
• func MessageComponents(agentToServer *protobufs.AgentToServer) []string
• func SyncOne[T protoiface.MessageV1](ctx context.Context, logger *zap.Logger, ...) (updated bool)
• func UpdateAgentStatus(logger *zap.Logger, agent *model.Agent, ...)
• func UpdateOpAmpAgentDetails(agent *model.Agent, conn opamp.Connection, desc *protobufs.AgentDescription)
• type AgentConnectionState
• type AgentState
  • func DecodeState(state interface{}) (*AgentState, error)
  • func (s *AgentState) Configuration() *observiq.RawAgentConfiguration
  • func (s *AgentState) IsMissingMessage(agentToServer *protobufs.AgentToServer) bool
  • func (s *AgentState) UpdateSequenceNumber(agentToServer *protobufs.AgentToServer)
• type Connection
• type Connections
• type SerializedAgentState
  • func EncodeState(state *AgentState) *SerializedAgentState
  • func (s *SerializedAgentState) Scan(value interface{}) error
  • func (s SerializedAgentState) Value() (driver.Value, error)

Constants

const CollectorPackageName = "observiq-otel-collector"

CollectorPackageName is the name for the top level packages for this collector

Variables

var (
	SyncAgentDescription   = agentDescriptionSyncer{}
	SyncEffectiveConfig    = effectiveConfigSyncer{}
	SyncRemoteConfigStatus = remoteConfigStatusSyncer{}
	SyncPackageStatuses    = packageStatusesSyncer{}
)

These are the functions that are used to update the agent state from the OpAMP messages

var ErrAgentNotRegistered = errors.New("agent not registered")

ErrAgentNotRegistered is returned when an agent that is not registered sends a message

Functions

func MessageComponents

func MessageComponents(agentToServer *protobufs.AgentToServer) []string

MessageComponents returns the names of the components in the message

func SyncOne

func SyncOne[T protoiface.MessageV1](ctx context.Context, logger *zap.Logger, agentToServer *protobufs.AgentToServer, state *AgentState, conn opamp.Connection, agent *model.Agent, response *protobufs.ServerToAgent, syncer messageSyncer[T]) (updated bool)

SyncOne syncs a single message

func UpdateAgentStatus

func UpdateAgentStatus(logger *zap.Logger, agent *model.Agent, remoteStatus *protobufs.RemoteConfigStatus)

UpdateAgentStatus modifies the agent status based on the RemoteConfigStatus, if available

func UpdateOpAmpAgentDetails

func UpdateOpAmpAgentDetails(agent *model.Agent, conn opamp.Connection, desc *protobufs.AgentDescription)

UpdateOpAmpAgentDetails updates the agent details from the opamp agent description.

Types

type AgentConnectionState

type AgentConnectionState struct {
	AgentID  string
	Conn     opamp.Connection
	SendLock sync.Mutex
}

AgentConnectionState is the state of an agent connection

type AgentState

type AgentState struct {
	SequenceNum uint64                  `json:"sequenceNum" yaml:"sequenceNum" mapstructure:"sequenceNum"`
	Status      protobufs.AgentToServer `json:"status,omitempty" yaml:"status,omitempty" mapstructure:"status"`
}

AgentState stores OpAMP state for the agent

func DecodeState

func DecodeState(state interface{}) (*AgentState, error)

DecodeState decodes the agent state from a serialized form

func (*AgentState) Configuration

func (s *AgentState) Configuration() *observiq.RawAgentConfiguration

Configuration returns the current configuration

func (*AgentState) IsMissingMessage

func (s *AgentState) IsMissingMessage(agentToServer *protobufs.AgentToServer) bool

IsMissingMessage returns true if the message is missing

func (*AgentState) UpdateSequenceNumber

func (s *AgentState) UpdateSequenceNumber(agentToServer *protobufs.AgentToServer)

UpdateSequenceNumber updates the sequence number

type Connection

type Connection interface {
	opamp.Connection
}

Connection is an interface that wraps the opamp.Connection interface

type Connections

type Connections[S any] interface {
	// Connected returns true if the agent with the specified agentID is connected
	Connected(agentID string) bool

	// ConnectedAgentIDs returns the IDs of all connected agents
	ConnectedAgentIDs(context.Context) []string

	// ConnectedAgentsCount returns the number of connected agents
	ConnectedAgentsCount(context.Context) int

	// OnConnecting should be called within the opamp OnConnecting callback. It adds the association of the accountID to
	// the agentID after it has been determined. The associated opamp.Connection is not yet available. It will be added in
	// OnMessage.
	OnConnecting(ctx context.Context, agentID string) S

	// OnMessage should be called within the opamp OnMessage callback. It adds the association of the opamp.Connection to
	// the connectionState.
	OnMessage(agentID string, conn opamp.Connection) (S, error)

	// OnConnectionClose removes the connection and returns the state and the count of remaining agents connected.
	OnConnectionClose(conn opamp.Connection) (state S, count int)

	// StateForAgentID returns the current state for the specified agentID or nil if there is no connection
	StateForAgentID(agentID string) S

	// StateForConnection returns the current state for the specified [opamp.Connection] or nil if there is no connection
	StateForConnection(opamp.Connection) S
}

Connections is an interface to manage agent connections and manage per-connection state. The generic type S can be specified, but AgentConnectionState is a good default

type SerializedAgentState

type SerializedAgentState struct {
	SequenceNum uint64 `json:"sequenceNum" yaml:"sequenceNum" mapstructure:"sequenceNum"`
	Status      string `json:"status,omitempty" yaml:"status,omitempty" mapstructure:"status"`
}

SerializedAgentState is stored on the model.Agent in a partially serialized form. The status is a base64-encoded protobuf.

func EncodeState

func EncodeState(state *AgentState) *SerializedAgentState

EncodeState encodes the agent state into a serialized form

func (*SerializedAgentState) Scan

func (s *SerializedAgentState) Scan(value interface{}) error

Scan is used to translate from a JSONB field in postgres to serializedAgentState

func (SerializedAgentState) Value

func (s SerializedAgentState) Value() (driver.Value, error)

Value is used to translate to a JSONB field for postgres storage