telemetry

package
v0.10.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 19, 2026 License: Apache-2.0 Imports: 36 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	RuntimeProcessGraphVersion = "agentprov.runtime_process/v3"
	RuntimeProcessGraphV2      = "agentprov.runtime_process/v2"
)
View Source 
const AuditAssociationRule = "agentprov.audit_lifecycle_operation/v2"
View Source 
const AuditCollectionSchema = "agentprovenance.audit_collection/v1"
View Source 
const AuditSchemaVersion = "agentprovenance.access_control_audit/v1"
View Source 
const MaxAuditLineBytes = 64 << 10
View Source 
const NativeRejectionLimit = 64
View Source 
const ProducerHealthSchemaVersion = "agentprovenance.producer_health/v1"

Variables

View Source 
var ErrAuditAssociationQuery = errors.New("invalid audit association query")

Functions

func AllowedEventType 

func AllowedEventType(eventType string) bool

func AuditAssociationsHandler added in v0.10.0 

func AuditAssociationsHandler(db *sql.DB) http.Handler

func CorrelationClass added in v0.2.0 

func CorrelationClass(source, method, containerID string, confidence float64) string

CorrelationClass labels the PROVENANCE of an event's correlation so a security consumer can distinguish a self-join from independent corroboration. It is the correlation-STRENGTH axis only; whether the scope launched the process is a separate, orthogonal fact (see SelfLaunched). The two combine in the UI: a sensor event matched to a self-launched binding is kernel_correlated AND self_launched. It is a pure function of already-stored fields:

  • self_observed: we only know about this event because AgentProvenance itself produced the observation (record's own process-tree sampling / file diff / direct record events, or the zero_sdk_process_tree method). There is no independent kernel witness, so its confidence is self-consistency, NOT independent evidence. Keyed on the event SOURCE, not on any synthetic container-id string: a real kernel event that merely matched a record-launched binding is independently witnessed and stays kernel_correlated (that is what SelfLaunched is for).
  • context_asserted: the caller provided full run/session/tool_call context; no correlation was performed.
  • kernel_correlated: independent system telemetry (Falco/Tetragon/own eBPF sensor) joined to app context through a binding. This is the real claim.
  • uncorrelated: could not be resolved.

func IngestFiltered 

func IngestFiltered(db *sql.DB, event IngestEvent) (string, error)

func NativeStreamRunning added in v0.8.0 

func NativeStreamRunning(paths store.Paths) (bool, error)

NativeStreamRunning tests the process-held collector lock; persisted probe reports alone are historical snapshots and cannot establish liveness.

func RebuildEventWindows 

func RebuildEventWindows(db *sql.DB, runID string) (int, error)

func SelfLaunched added in v0.5.0 

func SelfLaunched(source, bindingSource string) bool

SelfLaunched reports whether the process behind this event was started by AgentProvenance itself (zero-SDK record mode / control-plane launch). It is orthogonal to CorrelationClass: an independently-witnessed kernel event can be self_launched too. It is derived from the event source (record's own events are self-launched by construction) and from the binding_source of the binding the event resolved against, so it survives onto agentprov_ebpf sensor events that matched a record/control-plane binding.

func TelemetrySource 

func TelemetrySource(source string, correlationMethod string) bool

func ValidateRawPayload 

func ValidateRawPayload(eventType, payload string) error

func ValidateStoredPayload 

func ValidateStoredPayload(eventType, payload string) error

Types

type AuditAssociation added in v0.10.0 

type AuditAssociation struct {
	Status              string           `json:"status"`
	Basis               string           `json:"basis,omitempty"`
	Reason              string           `json:"reason,omitempty"`
	CandidateCount      int              `json:"candidate_count"`
	CandidatesTruncated bool             `json:"candidates_truncated,omitempty"`
	Candidates          []AuditCandidate `json:"candidates,omitempty"`
}

type AuditAssociationItem added in v0.10.0 

type AuditAssociationItem struct {
	EventID       string           `json:"event_id"`
	Source        AuditSource      `json:"source"`
	TimeWindow    *AuditTimeWindow `json:"time_window,omitempty"`
	Mechanism     string           `json:"mechanism"`
	Decision      string           `json:"decision"`
	Process       AuditAssociation `json:"process"`
	Operation     AuditAssociation `json:"operation"`
	ProcessExits  []string         `json:"process_exit_refs,omitempty"`
	Operations    []AuditOperation `json:"operations,omitempty"`
	EvidenceScope string           `json:"evidence_scope"`
}

type AuditAssociationOptions added in v0.10.0 

type AuditAssociationOptions struct {
	RunID   string
	EventID string
	Limit   int
	Cursor  string
}

func ParseAuditAssociationQuery added in v0.10.0 

func ParseAuditAssociationQuery(q url.Values) (AuditAssociationOptions, error)

ParseAuditAssociationQuery is shared by HTTP and tool callers. Reject unknown filters instead of silently returning evidence outside the requested scope.

type AuditAssociationReport added in v0.10.0 

type AuditAssociationReport struct {
	SchemaVersion string                 `json:"schema_version"`
	Rule          string                 `json:"rule"`
	RunID         string                 `json:"run_id"`
	Status        string                 `json:"status"`
	Limit         int                    `json:"limit"`
	NextCursor    string                 `json:"next_cursor,omitempty"`
	HasMore       bool                   `json:"has_more"`
	RuntimeRows   int                    `json:"runtime_rows_examined"`
	AnalysisScope string                 `json:"analysis_scope"`
	Issues        []string               `json:"issues,omitempty"`
	Items         []AuditAssociationItem `json:"items"`
}

Association is a read-only projection. It does not change the audit record's PID, binding or Session, and may change when late evidence adds candidates.

func BuildAuditAssociations added in v0.10.0 

func BuildAuditAssociations(ctx context.Context, db *sql.DB, opts AuditAssociationOptions) (AuditAssociationReport, error)

type AuditCandidate added in v0.10.0 

type AuditCandidate struct {
	ID           string   `json:"id"`
	EvidenceRefs []string `json:"evidence_refs"`
}

type AuditCollectionObservation added in v0.10.0 

type AuditCollectionObservation struct {
	SchemaVersion      string `json:"schema_version"`
	EvidenceKind       string `json:"evidence_kind"`
	ReaderID           string `json:"reader_id"`
	SourcePath         string `json:"source_path"`
	Mechanism          string `json:"mechanism"`
	RawRecordsRetained bool   `json:"raw_records_retained"`
	Status             string `json:"status"`
	Reason             string `json:"reason"`
	CoverageScope      string `json:"coverage_scope"`
	UpstreamLoss       string `json:"upstream_loss"`
	ObservedAt         string `json:"observed_at"`
	LastStartAt        string `json:"last_start_at"`
	LastReadAt         string `json:"last_read_at,omitempty"`
	CommittedOffset    int64  `json:"committed_offset"`
	PendingBytes       int64  `json:"pending_bytes"`
	Read               int    `json:"read"`
	Ingested           int    `json:"ingested"`
	Duplicates         int    `json:"duplicates"`
	Skipped            int    `json:"skipped"`
	Failed             int    `json:"failed"`
	IdentityIncomplete int    `json:"identity_incomplete"`
	Batches            int    `json:"batches"`
	LastBatchID        string `json:"last_batch_id,omitempty"`
}

These counters describe the selected source file, not every event the kernel could have audited. Unknown upstream loss is deliberately not a zero counter.

type AuditContextRef added in v0.10.0 

type AuditContextRef struct {
	EventID    string  `json:"event_id"`
	SessionID  string  `json:"session_id,omitempty"`
	ToolCallID string  `json:"tool_call_id,omitempty"`
	Method     string  `json:"method"`
	Confidence float64 `json:"confidence"`
}

Context is cited from runtime records, not inferred from a shared PID/cgroup.

type AuditFollowOptions added in v0.10.0 

type AuditFollowOptions struct {
	AuditIngestOptions
	ReaderID     string
	PollInterval time.Duration
	Once         bool
}

type AuditFollowStatus added in v0.10.0 

type AuditFollowStatus struct {
	SchemaVersion string                     `json:"schema_version"`
	RunID         string                     `json:"run_id"`
	Reader        AuditCollectionObservation `json:"reader"`
}

func FollowAuditFile added in v0.10.0 

func FollowAuditFile(ctx context.Context, db *sql.DB, paths store.Paths, opts AuditFollowOptions, report func(AuditFollowStatus) error) (AuditFollowStatus, error)

FollowAuditFile reads existing append-only audit text or journal JSON files. It uses the source as its durable queue, bounds each in-memory batch, and commits events, receipts and the source checkpoint together. It stops on rotation/truncation rather than guessing where an unread generation went.

type AuditIngestOptions added in v0.10.0 

type AuditIngestOptions struct {
	RunID     string
	Path      string
	Source    AuditSource
	Mechanism string // all, apparmor, or seccomp
	RetainRaw bool   // caller resolves the run's fixed capture policy
}

type AuditIngestResult added in v0.10.0 

type AuditIngestResult struct {
	SchemaVersion      string      `json:"schema_version"`
	RunID              string      `json:"run_id"`
	Path               string      `json:"path"`
	Status             string      `json:"status"`
	RawRecordsRetained bool        `json:"raw_records_retained"`
	Read               int         `json:"read"`
	Ingested           int         `json:"ingested"`
	Duplicates         int         `json:"duplicates"`
	Skipped            int         `json:"skipped"`
	Failed             int         `json:"failed"`
	IdentityIncomplete int         `json:"identity_incomplete"`
	Batches            int         `json:"batches"`
	LastBatchID        string      `json:"last_batch_id,omitempty"`
	FileSHA256         string      `json:"file_sha256,omitempty"`
	Rows               []RowResult `json:"row_results,omitempty"`
	RowsTruncated      bool        `json:"row_results_truncated,omitempty"`
	Errors             []string    `json:"errors,omitempty"`
	ErrorsTruncated    bool        `json:"errors_truncated,omitempty"`
}

func IngestAuditReader added in v0.10.0 

func IngestAuditReader(ctx context.Context, db *sql.DB, opts AuditIngestOptions, input io.Reader) (AuditIngestResult, error)

IngestAuditReader imports existing logs without changing auditd, AppArmor or seccomp. Batches are bounded and transactional, including dedup and edges. Re-running a failed import is safe when source host/boot identity is present.

type AuditOperation added in v0.10.0 

type AuditOperation struct {
	ID              string            `json:"id"`
	ProcessInstance string            `json:"process_instance_id"`
	ThreadInstance  string            `json:"thread_instance_id"`
	Kind            string            `json:"kind"`
	AttemptState    string            `json:"attempt_state"`
	ResultState     string            `json:"result_state"`
	AttemptCount    int               `json:"attempt_record_count"`
	ResultCount     int               `json:"result_record_count"`
	RefsTruncated   bool              `json:"refs_truncated,omitempty"`
	AttemptRefs     []string          `json:"attempt_refs,omitempty"`
	ResultRefs      []string          `json:"result_refs,omitempty"`
	Result          *int64            `json:"syscall_result,omitempty"`
	Outcome         string            `json:"syscall_outcome,omitempty"`
	Context         []AuditContextRef `json:"context,omitempty"`
}

type AuditRawRecord added in v0.10.0 

type AuditRawRecord struct {
	State       string `json:"state"`
	SHA256      string `json:"sha256"`
	Bytes       int    `json:"bytes"`
	Text        string `json:"text,omitempty"`
	Reason      string `json:"reason,omitempty"`
	DigestScope string `json:"digest_scope"`
	Redacted    bool   `json:"redacted"`
}

type AuditRecord added in v0.10.0 

type AuditRecord struct {
	SchemaVersion        string `json:"schema_version"`
	Mechanism            string `json:"mechanism"`
	EvidenceKind         string `json:"evidence_kind"`
	RecordType           string `json:"record_type"`
	AuditID              string `json:"audit_id"`
	Timestamp            string `json:"timestamp"`
	TimestampPrecisionNS int64  `json:"timestamp_precision_ns"`
	AuditSource
	IdentitySource   string         `json:"identity_source"`
	SourceFormat     string         `json:"source_format"`
	JournalCursor    string         `json:"journal_cursor,omitempty"`
	RecordID         string         `json:"record_id,omitempty"`
	Decision         string         `json:"decision"`
	AuditPID         *int64         `json:"audit_pid,omitempty"`
	PIDSemantics     string         `json:"pid_semantics"`
	Comm             string         `json:"comm,omitempty"`
	Executable       string         `json:"executable,omitempty"`
	Profile          string         `json:"profile,omitempty"`
	ProfileNamespace string         `json:"profile_namespace,omitempty"`
	Label            string         `json:"label,omitempty"`
	Class            string         `json:"class,omitempty"`
	Operation        string         `json:"operation,omitempty"`
	Target           string         `json:"target,omitempty"`
	RequestedMask    string         `json:"requested_mask,omitempty"`
	DeniedMask       string         `json:"denied_mask,omitempty"`
	Arch             string         `json:"arch,omitempty"`
	ABI              string         `json:"abi,omitempty"`
	SyscallNumber    *int64         `json:"syscall_number,omitempty"`
	SyscallName      string         `json:"syscall_name,omitempty"`
	ActionCode       *uint32        `json:"action_code,omitempty"`
	ActionData       *uint16        `json:"action_data,omitempty"`
	Signal           *int64         `json:"signal,omitempty"`
	Warnings         []string       `json:"warnings,omitempty"`
	RawRecord        AuditRawRecord `json:"raw_record"`
}

func ParseAuditLine added in v0.10.0 

func ParseAuditLine(line []byte, source AuditSource) (AuditRecord, bool, error)

ParseAuditLine accepts raw audit records (including kernel/syslog prefixes) and journalctl JSON. It does not read /proc, install policies, or infer a process lifetime from a logged PID. Unrelated log records are skipped.

type AuditSource added in v0.10.0 

type AuditSource struct {
	HostID string `json:"host_id,omitempty"`
	BootID string `json:"boot_id,omitempty"`
}

Source identities describe the supplied log, never the machine doing import.

type AuditTimeWindow added in v0.10.0 

type AuditTimeWindow struct {
	From                  string `json:"from"`
	Until                 string `json:"until"`
	DecimalPrecisionNS    int64  `json:"decimal_precision_ns"`
	CoarseClockResolution int64  `json:"source_coarse_clock_resolution_ns"`
	Basis                 string `json:"basis"`
}

type BatchRecord 

type BatchRecord struct {
	ID             string `json:"id"`
	RunID          string `json:"run_id"`
	Format         string `json:"format"`
	Path           string `json:"path"`
	FileSHA256     string `json:"file_sha256"`
	Read           int    `json:"read"`
	Ingested       int    `json:"ingested"`
	Skipped        int    `json:"skipped"`
	Failed         int    `json:"failed"`
	EventIDsSHA256 string `json:"event_ids_sha256"`
	CreatedAt      string `json:"created_at"`
}

func ListBatches 

func ListBatches(db *sql.DB, runID string) ([]BatchRecord, error)

type CorrelationBinding 

type CorrelationBinding struct {
	ID            string   `json:"id"`
	RunID         string   `json:"run_id,omitempty"`
	SessionID     string   `json:"session_id,omitempty"`
	AttemptID     string   `json:"attempt_id,omitempty"`
	ToolCallID    string   `json:"tool_call_id,omitempty"`
	ProcessID     string   `json:"process_id,omitempty"`
	ContainerID   string   `json:"container_id,omitempty"`
	CgroupID      string   `json:"cgroup_id,omitempty"`
	RootPID       int64    `json:"root_pid,omitempty"`
	PID           int64    `json:"pid,omitempty"`
	StartedAt     string   `json:"started_at,omitempty"`
	EndedAt       string   `json:"ended_at,omitempty"`
	BindingSource string   `json:"binding_source,omitempty"`
	Confidence    float64  `json:"confidence"`
	IdentityKeys  []string `json:"identity_keys,omitempty"`
}

type CorrelationContext 

type CorrelationContext struct {
	RunID      string `json:"run_id,omitempty"`
	SessionID  string `json:"session_id,omitempty"`
	AttemptID  string `json:"attempt_id,omitempty"`
	ToolCallID string `json:"tool_call_id,omitempty"`
	ProcessID  string `json:"process_id,omitempty"`
	SnapshotID string `json:"snapshot_id,omitempty"`
}

type CorrelationEvent 

type CorrelationEvent struct {
	ID         string `json:"id"`
	Type       string `json:"type"`
	Source     string `json:"source"`
	RawEventID string `json:"raw_event_id,omitempty"`
	CreatedAt  string `json:"created_at"`
}

type CorrelationEvidence 

type CorrelationEvidence struct {
	Event           CorrelationEvent    `json:"event"`
	RawIdentity     CorrelationIdentity `json:"raw_identity"`
	ResolvedContext CorrelationContext  `json:"resolved_context"`
	Binding         *CorrelationBinding `json:"binding,omitempty"`
	Match           CorrelationMatch    `json:"match"`
	Drilldowns      []string            `json:"drilldowns,omitempty"`
}

type CorrelationIdentity 

type CorrelationIdentity struct {
	RunID       string   `json:"run_id,omitempty"`
	ProcessID   string   `json:"process_id,omitempty"`
	ContainerID string   `json:"container_id,omitempty"`
	CgroupID    string   `json:"cgroup_id,omitempty"`
	PID         int64    `json:"pid,omitempty"`
	TGID        int64    `json:"tgid,omitempty"`
	PPID        int64    `json:"ppid,omitempty"`
	Keys        []string `json:"keys,omitempty"`
}

type CorrelationMatch 

type CorrelationMatch struct {
	Status              string   `json:"status"`
	Method              string   `json:"method"`
	Confidence          float64  `json:"confidence"`
	BindingID           string   `json:"binding_id,omitempty"`
	MatchedKeys         []string `json:"matched_keys,omitempty"`
	Reason              string   `json:"reason"`
	TimeWindow          string   `json:"time_window,omitempty"`
	ScopeOnly           bool     `json:"scope_only,omitempty"`
	Ambiguous           bool     `json:"ambiguous,omitempty"`
	CandidateBindingIDs []string `json:"candidate_binding_ids,omitempty"`
	CandidateCount      int      `json:"candidate_count,omitempty"`
}

type CorrelationReport 

type CorrelationReport struct {
	SchemaVersion string                `json:"schema_version"`
	RunID         string                `json:"run_id,omitempty"`
	EventID       string                `json:"event_id,omitempty"`
	ResultSetID   string                `json:"result_set_id"`
	PageHash      string                `json:"page_hash"`
	Count         int                   `json:"count"`
	Items         []CorrelationEvidence `json:"correlations"`
}

func BuildCorrelationReport 

func BuildCorrelationReport(db *sql.DB, opts CorrelationReportOptions) (CorrelationReport, error)

type CorrelationReportOptions 

type CorrelationReportOptions struct {
	RunID   string
	EventID string
}

type EventExplanation 

type EventExplanation struct {
	Receiver              string   `json:"receiver"`
	SourceFormat          string   `json:"source_format"`
	RawEventID            string   `json:"raw_event_id,omitempty"`
	NormalizedEventType   string   `json:"normalized_event_type"`
	SchemaStatus          string   `json:"schema_status"`
	SchemaError           string   `json:"schema_error,omitempty"`
	IdentityKeys          []string `json:"identity_keys,omitempty"`
	CorrelationMethod     string   `json:"correlation_method,omitempty"`
	CorrelationConfidence float64  `json:"correlation_confidence"`
	CorrelationClass      string   `json:"correlation_class,omitempty"`
	CorrelationStatus     string   `json:"correlation_status"`
}

func ExplainEventRecord 

func ExplainEventRecord(event EventRecord) EventExplanation

type EventRecord 

type EventRecord struct {
	ID                    string  `json:"id"`
	RunID                 string  `json:"run_id"`
	SessionID             string  `json:"session_id"`
	ToolCallID            string  `json:"tool_call_id"`
	ProcessID             string  `json:"process_id"`
	SnapshotID            string  `json:"snapshot_id"`
	RawEventID            string  `json:"raw_event_id"`
	CorrelationMethod     string  `json:"correlation_method"`
	CorrelationConfidence float64 `json:"correlation_confidence"`
	CorrelationClass      string  `json:"correlation_class,omitempty"`
	BindingSource         string  `json:"binding_source,omitempty"`
	SelfLaunched          bool    `json:"self_launched,omitempty"`
	ContainerID           string  `json:"container_id"`
	CgroupID              string  `json:"cgroup_id"`
	PID                   int64   `json:"pid"`
	TGID                  int64   `json:"tgid"`
	PPID                  int64   `json:"ppid"`
	Source                string  `json:"source"`
	EventType             string  `json:"event_type"`
	Payload               string  `json:"payload"`
	CreatedAt             string  `json:"created_at"`
}

func ListEvents 

func ListEvents(db *sql.DB, runID, sessionID string) ([]EventRecord, error)

func ListEventsFiltered 

func ListEventsFiltered(db *sql.DB, filter Filter) ([]EventRecord, error)

type EventWindow 

type EventWindow struct {
	RunID           string `json:"run_id"`
	SessionID       string `json:"session_id,omitempty"`
	ToolCallID      string `json:"tool_call_id,omitempty"`
	Source          string `json:"source"`
	EventType       string `json:"event_type"`
	WindowSeconds   int    `json:"window_seconds"`
	WindowStart     string `json:"window_start"`
	EventCount      int    `json:"event_count"`
	ResolvedCount   int    `json:"resolved_count"`
	UnresolvedCount int    `json:"unresolved_count"`
	HighRiskCount   int    `json:"high_risk_count"`
	UpdatedAt       string `json:"updated_at"`
}

type EventWindowFilter 

type EventWindowFilter struct {
	RunID         string `json:"run_id,omitempty"`
	SessionID     string `json:"session_id,omitempty"`
	ToolCallID    string `json:"tool_call_id,omitempty"`
	Type          string `json:"type,omitempty"`
	Source        string `json:"source,omitempty"`
	WindowSeconds int    `json:"window_seconds,omitempty"`
}

type EventWindowResult 

type EventWindowResult struct {
	SchemaVersion string            `json:"schema_version"`
	Filter        EventWindowFilter `json:"filter"`
	WindowCount   int               `json:"window_count"`
	ResultSetID   string            `json:"result_set_id"`
	PageHash      string            `json:"page_hash"`
	Windows       []EventWindow     `json:"windows"`
}

func ListEventWindows 

func ListEventWindows(db *sql.DB, filter EventWindowFilter) (EventWindowResult, error)

type FalcoIngestOptions 

type FalcoIngestOptions struct {
	Path       string
	RunID      string
	SessionID  string
	AttemptID  string
	ToolCallID string
	ProcessID  string
	SnapshotID string
	RolloutID  string
}

type Filter 

type Filter struct {
	RunID      string
	SessionID  string
	Type       string
	ToolCallID string
}

type IngestEvent 

type IngestEvent struct {
	RunID       string
	RolloutID   string
	AttemptID   string
	SessionID   string
	ToolCallID  string
	ProcessID   string
	SnapshotID  string
	RawEventID  string
	ContainerID string
	CgroupID    string
	PID         int64
	TGID        int64
	PPID        int64
	Timestamp   string
	Source      string
	EventType   string
	Payload     string
	// BindingSource records which binding tier produced this event's
	// correlation (e.g. ai_asserted, zero_sdk_record, control_plane). When the
	// event is resolved against a stored binding it is overwritten from the
	// matched binding; callers with directly-provided context may set it so the
	// self-launched dimension is preserved on the provided_context path.
	BindingSource string
}

type JSONLIngestOptions 

type JSONLIngestOptions struct {
	Format     string
	Path       string
	RunID      string
	SessionID  string
	AttemptID  string
	ToolCallID string
	ProcessID  string
	SnapshotID string
	RolloutID  string
	// ExcludePathPrefixes drops file-oriented events whose path is under any of
	// these prefixes. Used to keep the streaming sensor from observing (and
	// ingesting, and thereby generating more of) AgentProvenance's OWN I/O --
	// its data-dir snapshot copies and store DB writes -- which otherwise forms
	// a self-feedback storm that buries and drops real scope events.
	ExcludePathPrefixes []string
	// ExcludeCgroupIDs drops events whose cgroup id matches -- typically the
	// supervisor's own cgroup, so the sensor never ingests its own process's
	// activity. A second, cgroup-granular guard alongside ExcludePathPrefixes.
	ExcludeCgroupIDs []string
	// DropUncorrelated, when set, discards events that resolve to no tracked
	// scope (method unresolved / confidence 0). The always-on streaming sensor
	// sees host-wide activity; events belonging to no ToolCallScope are host
	// noise. Persisting them pollutes the store and -- because the ingest batch
	// references them -- fails per-run graph verify. Dropping them keeps a
	// capture clean and verifiable. Off by default (a per-node security
	// deployment may want unattributed activity retained).
	DropUncorrelated bool
}

type JSONLIngestResult 

type JSONLIngestResult struct {
	BatchID           string          `json:"batch_id,omitempty"`
	Format            string          `json:"format"`
	Path              string          `json:"path"`
	FileSHA256        string          `json:"file_sha256,omitempty"`
	Read              int             `json:"read"`
	Ingested          int             `json:"ingested"`
	Skipped           int             `json:"skipped"`
	Failed            int             `json:"failed"`
	EventIDs          []string        `json:"event_ids,omitempty"`
	EventIDsSHA256    string          `json:"event_ids_sha256,omitempty"`
	Excluded          int             `json:"excluded,omitempty"`
	Dropped           int             `json:"dropped_uncorrelated,omitempty"`
	PolicyDecisions   int             `json:"policy_decisions"`
	PolicyDecisionIDs []string        `json:"policy_decision_ids,omitempty"`
	ReceiverSummary   ReceiverSummary `json:"receiver_summary"`
	Rows              []RowResult     `json:"row_results,omitempty"`
	RowsTruncated     bool            `json:"row_results_truncated,omitempty"`
	Errors            []string        `json:"errors,omitempty"`
}

func IngestFalco 

func IngestFalco(db *sql.DB, opts FalcoIngestOptions, input io.Reader) (JSONLIngestResult, error)

func IngestJSONL 

func IngestJSONL(db *sql.DB, opts JSONLIngestOptions) (JSONLIngestResult, error)

func IngestJSONLReader added in v0.2.0 

func IngestJSONLReader(db *sql.DB, opts JSONLIngestOptions, input io.Reader) (JSONLIngestResult, error)

IngestJSONLReader ingests JSONL telemetry from any reader, so the agentprov sensor (or any substrate) can pipe its stdout straight into the receiver (`--file -`) instead of staging a file first. FileSHA256 is computed over the exact bytes streamed, so a pipe and the equivalent saved file hash identically.

type ListOptions 

type ListOptions struct {
	Filter Filter
	Limit  int
	Cursor string
}

type ListResult 

type ListResult struct {
	SchemaVersion string        `json:"schema_version"`
	Filter        Filter        `json:"filter"`
	Limit         int           `json:"limit"`
	Cursor        string        `json:"cursor,omitempty"`
	NextCursor    string        `json:"next_cursor,omitempty"`
	HasMore       bool          `json:"has_more"`
	EventCount    int           `json:"event_count"`
	TotalCount    int           `json:"total_count"`
	ResultSetID   string        `json:"result_set_id"`
	PageHash      string        `json:"page_hash"`
	Events        []EventRecord `json:"events"`
}

func ListEventsPage 

func ListEventsPage(db *sql.DB, opts ListOptions) (ListResult, error)

type NativeRejection added in v0.9.0 

type NativeRejection struct {
	Timestamp         string `json:"timestamp"`
	EventType         string `json:"event_type"`
	PID               int64  `json:"pid"`
	CgroupID          string `json:"cgroup_id"`
	ProcessInstanceID string `json:"process_instance_id,omitempty"`
	Reason            string `json:"reason"`
}

Rejection diagnostics contain identity and a bounded validation reason, never the rejected body, file contents or command arguments.

func ReadNativeRejections added in v0.9.0 

func ReadNativeRejections(logs string) ([]NativeRejection, error)

type NativeStream added in v0.8.0 

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

NativeStream accepts JSONL writes from the sensor, normalizes/redacts them before persistence, fsyncs accepted rows, and seals bounded spool batches. Only counters and one partial JSON line are retained in memory. A store lock prevents two collectors from reclaiming each other's in-flight batches.

func NewNativeStream added in v0.8.0 

func NewNativeStream(db *sql.DB, paths store.Paths, options NativeStreamOptions) (*NativeStream, error)

func (*NativeStream) Close added in v0.8.0 

func (n *NativeStream) Close() error

func (*NativeStream) Drain added in v0.9.0 

func (n *NativeStream) Drain(ctx context.Context) error

Drain retries sealed batches immediately after capture stops, including batches waiting for a late binding. Unrelated rows retain their original TTL; neither a full-node empty queue nor TTL expiry is required to seal a run.

func (*NativeStream) Flush added in v0.8.0 

func (n *NativeStream) Flush() error

Flush seals live batches. Processing is independent of capture; a slow database fills the bounded disk queue rather than accumulating event IDs.

func (*NativeStream) Process added in v0.8.0 

func (n *NativeStream) Process(limit int) error

func (*NativeStream) Run added in v0.8.0 

func (n *NativeStream) Run(ctx context.Context, report func(error)) error

Run retries all eligible batches fairly: pending batches wait one second, while new batches remain immediately eligible. A failed batch stays queued and its error/counter are durable; shutdown leaves pending rows for restart.

func (*NativeStream) Write added in v0.8.0 

func (n *NativeStream) Write(p []byte) (int, error)

type NativeStreamOptions added in v0.8.0 

type NativeStreamOptions struct {
	Ingest           JSONLIngestOptions
	BatchEvents      int
	BatchBytes       int64
	FlushInterval    time.Duration
	PendingTTL       time.Duration
	MaxQueuedBytes   int64
	MaxQueuedBatches int
	PolicyEnabled    bool
}

NativeStreamOptions bounds both resident memory and the on-disk backlog. PendingTTL starts at capture, not at replay, so restarts cannot renew it.

type NativeStreamStatus added in v0.8.0 

type NativeStreamStatus struct {
	Counters              map[string]int64 `json:"counters"`
	QueuedBatches         int              `json:"queued_batches"`
	QueuedBytes           int64            `json:"queued_bytes"`
	PendingEvents         int64            `json:"pending_events"`
	CleanupPendingBatches int              `json:"cleanup_pending_batches"`
	LastError             string           `json:"last_error,omitempty"`
}

func ReadNativeStreamStatus added in v0.8.0 

func ReadNativeStreamStatus(db *sql.DB) (NativeStreamStatus, error)

func ReadNativeStreamStatusContext added in v0.8.0 

func ReadNativeStreamStatusContext(ctx context.Context, db *sql.DB) (NativeStreamStatus, error)

type ProducerCoverageHealth added in v0.7.2 

type ProducerCoverageHealth struct {
	CorrelationRatio float64 `json:"correlation_ratio"`
	HasSensorDrops   bool    `json:"has_sensor_drops"`
	Complete         bool    `json:"complete"`
}

type ProducerEventHealth added in v0.7.2 

type ProducerEventHealth struct {
	RuntimeEvents       int            `json:"runtime_events"`
	CorrelatedEvents    int            `json:"correlated_events"`
	UncorrelatedEvents  int            `json:"uncorrelated_events"`
	SensorDroppedEvents int64          `json:"sensor_dropped_events"`
	BySource            map[string]int `json:"by_source"`
}

type ProducerHealthOptions added in v0.7.2 

type ProducerHealthOptions struct {
	RunID          string
	MaxQueued      int
	MaxQueuedBytes int64
	MaxBatchBytes  int64
	DropPolicy     string
}

type ProducerHealthReport added in v0.7.2 

type ProducerHealthReport struct {
	SchemaVersion string                 `json:"schema_version"`
	GeneratedAt   string                 `json:"generated_at"`
	RunID         string                 `json:"run_id,omitempty"`
	Limits        ProducerLimits         `json:"limits"`
	Spool         ProducerSpoolHealth    `json:"spool"`
	Events        ProducerEventHealth    `json:"events"`
	Coverage      ProducerCoverageHealth `json:"coverage"`
	// NativeCapture is node-wide even when RunID filters the event view: an
	// unresolved or dropped row cannot honestly be assigned to a particular run.
	NativeCapture NativeStreamStatus `json:"native_node_capture"`
}

func BuildProducerHealth added in v0.7.2 

func BuildProducerHealth(db *sql.DB, opts ProducerHealthOptions) (ProducerHealthReport, error)

type ProducerLimits added in v0.7.2 

type ProducerLimits struct {
	MaxQueuedBatches int    `json:"max_queued_batches"`
	MaxQueuedBytes   int64  `json:"max_queued_bytes"`
	MaxBatchBytes    int64  `json:"max_batch_bytes"`
	DropPolicy       string `json:"drop_policy"`
}

type ProducerSpoolHealth added in v0.7.2 

type ProducerSpoolHealth struct {
	ReceivedBatches   int            `json:"received_batches"`
	QueuedBatches     int            `json:"queued_batches"`
	QueuedBytes       int64          `json:"queued_bytes"`
	ProcessingBatches int            `json:"processing_batches"`
	ProcessedBatches  int            `json:"processed_batches"`
	DroppedBatches    int            `json:"dropped_batches"`
	DroppedBytes      int64          `json:"dropped_bytes"`
	FailedBatches     int            `json:"failed_batches"`
	IngestedEvents    int            `json:"ingested_events"`
	FailedEvents      int            `json:"failed_events"`
	ByStatus          map[string]int `json:"by_status"`
}

type ReceiverSummary 

type ReceiverSummary struct {
	DetectedFormats map[string]int `json:"detected_formats,omitempty"`
	EventTypes      map[string]int `json:"event_types,omitempty"`
	IdentityKeys    map[string]int `json:"identity_keys,omitempty"`
	Resolved        int            `json:"resolved"`
	Unresolved      int            `json:"unresolved"`
	Skipped         int            `json:"skipped"`
	Failed          int            `json:"failed"`
}

type RetentionOptions 

type RetentionOptions struct {
	RunID     string        `json:"run_id,omitempty"`
	OlderThan time.Duration `json:"-"`
	Cutoff    string        `json:"cutoff,omitempty"`
	MaxDelete int           `json:"max_delete"`
}

type RetentionResult 

type RetentionResult struct {
	SchemaVersion string   `json:"schema_version"`
	RunID         string   `json:"run_id,omitempty"`
	Cutoff        string   `json:"cutoff"`
	MaxDelete     int      `json:"max_delete"`
	Scanned       int      `json:"scanned"`
	Deleted       int      `json:"deleted"`
	Protected     int      `json:"protected"`
	DeletedIDs    []string `json:"deleted_ids,omitempty"`
}

func PruneRawEvents 

func PruneRawEvents(db *sql.DB, opts RetentionOptions) (RetentionResult, error)

type RowResult 

type RowResult struct {
	Line              int      `json:"line"`
	Status            string   `json:"status"`
	DetectedFormat    string   `json:"detected_format,omitempty"`
	EventID           string   `json:"event_id,omitempty"`
	EventType         string   `json:"event_type,omitempty"`
	Source            string   `json:"source,omitempty"`
	RawEventID        string   `json:"raw_event_id,omitempty"`
	IdentityKeys      []string `json:"identity_keys,omitempty"`
	CorrelationMethod string   `json:"correlation_method,omitempty"`
	Error             string   `json:"error,omitempty"`
}

type RuntimeProcessIdentity added in v0.10.0 

type RuntimeProcessIdentity struct {
	Version     string
	Process     RuntimeProcessNode
	Parent      RuntimeProcessNode
	ThreadGroup RuntimeProcessNode
}

func ProcessGraphIdentity added in v0.10.0 

func ProcessGraphIdentity(event EventRecord) RuntimeProcessIdentity

ProcessGraphIdentity is shared by ingest, verification, and query projections. An observation-local node preserves a captured PID/PPID/TGID without assuming that the same number in another event denotes the same process lifetime.

type RuntimeProcessNode added in v0.10.0 

type RuntimeProcessNode struct {
	ID         string
	PID        int64
	InstanceID string
	HostID     string
	Basis      string
}

type SpoolBackpressureError 

type SpoolBackpressureError struct {
	Queued         int    `json:"queued"`
	QueuedBytes    int64  `json:"queued_bytes"`
	IncomingBytes  int64  `json:"incoming_bytes"`
	MaxQueued      int    `json:"max_queued"`
	MaxQueuedBytes int64  `json:"max_queued_bytes"`
	MaxBatchBytes  int64  `json:"max_batch_bytes"`
	Reason         string `json:"reason"`
}

func (SpoolBackpressureError) Error 

func (e SpoolBackpressureError) Error() string

type SpoolBatch 

type SpoolBatch struct {
	ID            string `json:"id"`
	RunID         string `json:"run_id"`
	Format        string `json:"format"`
	SourcePath    string `json:"source_path"`
	SpoolPath     string `json:"spool_path"`
	FileSHA256    string `json:"file_sha256"`
	SizeBytes     int64  `json:"size_bytes"`
	Status        string `json:"status"`
	Priority      int    `json:"priority"`
	Attempts      int    `json:"attempts"`
	PolicyEnabled bool   `json:"policy_enabled"`
	IngestBatchID string `json:"ingest_batch_id"`
	IngestedCount int    `json:"ingested_count"`
	FailedCount   int    `json:"failed_count"`
	Error         string `json:"error"`
	CreatedAt     string `json:"created_at"`
	UpdatedAt     string `json:"updated_at"`
	ProcessedAt   string `json:"processed_at"`
	DroppedAt     string `json:"dropped_at"`
	DropReason    string `json:"drop_reason"`
}

type SpoolEnqueueRequest 

type SpoolEnqueueRequest struct {
	Format         string
	RunID          string
	SourcePath     string
	PolicyEnabled  bool
	MaxQueued      int
	MaxQueuedBytes int64
	MaxBatchBytes  int64
	DropPolicy     string
}

type SpoolProcessResult 

type SpoolProcessResult struct {
	Processed int                   `json:"processed"`
	Failed    int                   `json:"failed"`
	Batches   []SpoolProcessedBatch `json:"batches,omitempty"`
	Errors    []string              `json:"errors,omitempty"`
}

type SpoolProcessedBatch 

type SpoolProcessedBatch struct {
	ID                  string `json:"id"`
	IngestBatchID       string `json:"ingest_batch_id"`
	IngestedCount       int    `json:"ingested_count"`
	FailedCount         int    `json:"failed_count"`
	RowResultsTruncated bool   `json:"row_results_truncated,omitempty"`
}

type SpoolQueueStats added in v0.7.2 

type SpoolQueueStats struct {
	QueuedBatches int   `json:"queued_batches"`
	QueuedBytes   int64 `json:"queued_bytes"`
}

type SpoolService 

type SpoolService struct {
	DB    *sql.DB
	Paths store.Paths
}

func (SpoolService) CountQueued 

func (s SpoolService) CountQueued() (int, error)

func (SpoolService) Enqueue 

func (s SpoolService) Enqueue(req SpoolEnqueueRequest) (SpoolBatch, error)

func (SpoolService) List 

func (s SpoolService) List(runID string) ([]SpoolBatch, error)

func (SpoolService) Process 

func (s SpoolService) Process(limit int) (SpoolProcessResult, error)

func (SpoolService) QueueStats added in v0.7.2 

func (s SpoolService) QueueStats() (SpoolQueueStats, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.