omnisignal

README

OmniSignal

Unified signal ingestion abstraction for operational intelligence.

Overview

omnisignal provides a unified interface for ingesting operational signals from various external systems (alerting, ticketing, security, monitoring). It follows the same architectural pattern as omnillm:

• Provider interface defines the contract for signal sources • Registry allows dynamic provider registration • Thick providers use official SDKs (PagerDuty, Jira, New Relic) • Thin providers use native HTTP for sources without Go SDKs

Installation

go get github.com/plexusone/omnisignal

Quick Start

import (
    "github.com/plexusone/omnisignal"
    _ "github.com/plexusone/omnisignal/provider/pagerduty" // Register PagerDuty
)

func main() {
    provider, err := omnisignal.New("pagerduty", omnisignal.Config{
        APIKey: os.Getenv("PAGERDUTY_API_KEY"),
    })
    if err != nil {
        log.Fatal(err)
    }
    defer provider.Close()

    // Fetch incidents from the last 24 hours
    signals, err := provider.Fetch(ctx, omnisignal.FetchOptions{
        Since: time.Now().Add(-24 * time.Hour),
    })
    if err != nil {
        log.Fatal(err)
    }

    for _, sig := range signals {
        fmt.Printf("Signal: %s - %s (%s)\n", sig.ID, sig.Summary, sig.Severity)
    }
}

Available Providers

Built-in Providers

Provider	Type	Import Path	SDK
PagerDuty	Alerting	omnisignal/provider/pagerduty	go-pagerduty
Jira	Ticketing	omnisignal/provider/jira	go-jira
Analyst	Market Intelligence	omnisignal/provider/analyst	none (thin)
Competitive	Market Intelligence	omnisignal/provider/competitive	none (thin)

External Providers (Thick)

Provider	Type	Import Path	SDK
New Relic	Monitoring	omni-newrelic/omnisignal	newrelic-client-go
Aha	Product Feedback	grokify/aha-studio/omnisignal	aha-go

Planned Providers

Provider	Type	Priority
Zendesk	Ticketing	P1
Datadog	Monitoring	P1
Opsgenie	Alerting	P1
ServiceNow	ITSM	P2
Snyk	Security	P2

Provider Interface

type Provider interface {
    // Name returns the provider identifier
    Name() string

    // Fetch retrieves signals matching the given options
    Fetch(ctx context.Context, opts FetchOptions) ([]signal.Signal, error)

    // Subscribe opens a real-time stream of signals
    Subscribe(ctx context.Context, opts SubscribeOptions) (<-chan signal.Signal, error)

    // Capabilities returns what this provider supports
    Capabilities() Capabilities

    // Close releases any resources
    Close() error
}

Configuration

type Config struct {
    APIKey    string            // Primary authentication credential
    APISecret string            // Secondary credential (if required)
    BaseURL   string            // Override default API endpoint
    Timeout   time.Duration     // Request timeout
    RetryMax  int               // Max retry attempts
    Options   map[string]any    // Provider-specific options
}

Provider-Specific Options

PagerDuty:

omnisignal.Config{
    APIKey: "your-api-key",
}

Jira:

omnisignal.Config{
    BaseURL:   "https://company.atlassian.net",
    APIKey:    "user@example.com",  // Username
    APISecret: "api-token",         // API token
    Options: map[string]any{
        "projects": []string{"INFRA", "SUPPORT"},
    },
}

New Relic (via omni-newrelic):

omnisignal.Config{
    APIKey: "NRAK-xxx",
    Options: map[string]any{
        "account_id": 12345,
        "region":     "US",
    },
}

Analyst (Gartner, Forrester, IDC — see docs):

omnisignal.Config{
    Options: map[string]any{
        "source": "gartner", // or "forrester", "idc", "custom"
    },
}

Competitive (win/loss and gaps from CRM — see docs):

omnisignal.Config{
    Options: map[string]any{
        "source":              "salesforce", // or "hubspot", "clari", "gong", "custom"
        "competitor_mappings": map[string]string{"Okta Inc": "competitor:okta"},
        omnisignal.OptCustomerMappings: map[string]string{"Acme Corp": "customer:acme-001"},
        omnisignal.OptMarketMappings:   map[string]string{"IAM": "market:identity-governance"},
    },
}

Config Helpers

Config provides typed accessors for reading provider Options:

func (c Config) GetOption(key string, defaultVal any) any
func (c Config) GetStringOption(key, defaultVal string) string
func (c Config) GetStringMap(key string) map[string]string

GetStringMap accepts both map[string]string and map[string]any (with string values), which makes it safe to use with config loaded from JSON/YAML as well as Go literals.

Well-known option keys carry cross-repo reference mappings from source system values (e.g., organization names) to MarketSpec typed refs (e.g., customer:acme-001):

Constant	Key	Maps
OptCustomerMappings	customer_mappings	Organization/account names → customer refs
OptCapabilityMappings	capability_mappings	Components/labels → capability refs
OptMarketMappings	market_mappings	Categories → market refs

Fetch Options

type FetchOptions struct {
    Since      time.Time         // Filter signals after this time
    Until      time.Time         // Filter signals before this time
    Limit      int               // Maximum signals to return
    Statuses   []string          // Filter by status
    Severities []string          // Filter by severity
    Filters    map[string]string // Provider-specific filters
}

Output Format

All providers normalize their data to signal-spec types:

type Signal struct {
    ID          string           // Unique signal identifier
    Type        Type             // support_ticket, alert, security_finding, etc.
    Status      Status           // new, processing, mapped, archived
    Source      SourceSystem     // Origin system details
    Domain      Domain           // Category/subcategory
    Severity    Severity         // critical, high, medium, low, info
    Summary     string           // Brief description
    Description string           // Full content
    Entities    []Entity         // Referenced system components
    ObservedAt  time.Time        // When signal was observed
    ReceivedAt  time.Time        // When signal was received
    Tags        []Tag            // User-defined labels
    Metadata    map[string]any   // Source-specific data
}

Creating a Custom Provider

package myprovider

import "github.com/plexusone/omnisignal"

func init() {
    omnisignal.Register("myprovider", NewProvider, omnisignal.PriorityThick)
}

func NewProvider(cfg omnisignal.Config) (omnisignal.Provider, error) {
    // Initialize your provider
    return &Provider{config: cfg}, nil
}

type Provider struct {
    config omnisignal.Config
}

func (p *Provider) Name() string { return "myprovider" }

func (p *Provider) Fetch(ctx context.Context, opts omnisignal.FetchOptions) ([]signal.Signal, error) {
    // Fetch from your source and normalize to signal.Signal
}

func (p *Provider) Subscribe(ctx context.Context, opts omnisignal.SubscribeOptions) (<-chan signal.Signal, error) {
    return nil, omnisignal.ErrNotSupported
}

func (p *Provider) Capabilities() omnisignal.Capabilities {
    return omnisignal.Capabilities{
        SupportsBatchFetch: true,
    }
}

func (p *Provider) Close() error { return nil }

Curated Signals

Some sources (e.g., Aha Ideas, analyst findings, competitive deals) already represent a single aggregated data point rather than a raw event stream. Mark these signals as curated so the consolidation pipeline skips clustering and maps them directly to root causes:

sig.Metadata[omnisignal.MetaCurated] = true

// Or use the helper to check:
if omnisignal.IsCurated(sig.Metadata) {
    // Skip clustering, map directly to a canonical signal
}

See Metadata Conventions for the full raw vs. curated model.

Metrics Engine

The metrics package computes derived scores from signal sets via a pluggable formula registry:

import "github.com/plexusone/omnisignal/metrics"

// Built-in formulas: frustration, momentum, reach, urgency
result, err := metrics.Compute(ctx, "frustration", signals, metrics.Options{
    Weights: map[string]float64{"support_ticket": 1.5},
})

// Or run every registered formula at once
results, errs := metrics.ComputeAll(ctx, signals, metrics.Options{})

Formula	Description
frustration	Weighted signal count multiplied by the age (in days) of the oldest signal
momentum	Count of signals observed within a trailing window (default 30 days)
reach	Count of distinct customer references across all signals
urgency	Sum of severity-weighted signal counts

Weight overrides and window size can be loaded from JSON config and merged:

cfg, err := metrics.LoadConfig("metrics.json")
merged := defaultCfg.Merge(cfg)
result, err := metrics.Compute(ctx, "urgency", signals, merged.ToOptions())

Custom formulas can be added via metrics.Register(metrics.NewFormula(name, description, computeFn)).

Consolidation Pipeline

The consolidate package groups related raw signals into canonical root causes through a 5-stage pipeline: embed → cluster → summarize → review → attach.

import "github.com/plexusone/omnisignal/consolidate"

pipeline := consolidate.NewPipeline(
    consolidate.WithEmbedder(consolidate.NewOmniLLMEmbedder(omnillmClient, consolidate.EmbedderConfig{
        Model: "text-embedding-3-small",
    })),
    consolidate.WithSummarizer(consolidate.NewLLMSummarizer(omnillmClient, consolidate.SummarizerConfig{
        Model: "gpt-4o-mini",
    })),
    consolidate.WithReviewer(consolidate.NewMemoryReviewer(consolidate.ReviewConfig{
        AutoApproveThreshold: 5,
    })),
    consolidate.WithSimilarityThreshold(0.85),
)

result, err := pipeline.Process(ctx, signals)
// result.RootCauses, result.Clusters, result.Attached, result.Stats

• Embed — Embedder generates vector embeddings via OmniLLM (OmniLLMEmbedder)
• Cluster — signals are grouped by cosine similarity (Clusterer, IncrementalClusterer)
• Summarize — an LLM generates a root cause title, description, and symptom patterns per cluster (LLMSummarizer)
• Review — optional human review queue with approve/reject and auto-approve threshold (MemoryReviewer)
• Attach — new signals are linked to existing root causes incrementally via Pipeline.Attach

Curated signals skip embedding and clustering and map directly to root causes, preserving their existing aggregation.

Related Packages

• signal-spec - Canonical signal data model
• signal - Operational intelligence platform
• omnillm - Unified LLM abstraction

License

MIT

Documentation

Overview

Package omnisignal provides a unified interface for ingesting operational signals from various external systems (alerting, ticketing, security, monitoring).

omnisignal follows the same architectural pattern as omnillm:

• Provider interface defines the contract for signal sources
• Registry allows dynamic provider registration
• Thick providers use official SDKs (PagerDuty, Datadog)
• Thin providers use native HTTP for sources without Go SDKs

Example usage:

import (
    "github.com/plexusone/omnisignal"
    _ "github.com/plexusone/omnisignal/provider/pagerduty" // Register PagerDuty
)

provider, err := omnisignal.New("pagerduty", omnisignal.Config{
    APIKey: os.Getenv("PAGERDUTY_API_KEY"),
})
signals, err := provider.Fetch(ctx, omnisignal.FetchOptions{
    Since: time.Now().Add(-24 * time.Hour),
})

Index

• Constants
• Variables
• func ClearRegistry()
• func GetPriority(name string) int
• func IsCurated(metadata map[string]any) bool
• func IsRegistered(name string) bool
• func List() []string
• func Register(name string, factory ProviderFactory, priority int)
• func Unregister(name string)
• func WrapErrorByMessage(err error, context string) error
• func WrapHTTPError(err error, resp *http.Response, context string) error
• type Capabilities
• type Config
  • func (c Config) GetOption(key string, defaultVal any) any
  • func (c Config) GetStringMap(key string) map[string]string
  • func (c Config) GetStringOption(key, defaultVal string) string
• type FetchOptions
• type Provider
  • func MustNew(name string, cfg Config) Provider
  • func New(name string, cfg Config) (Provider, error)
• type ProviderFactory
• type SubscribeOptions

Constants

const (
	// OptCustomerMappings maps organization/account names to customer refs.
	OptCustomerMappings = "customer_mappings"
	// OptCapabilityMappings maps components/labels to capability refs.
	OptCapabilityMappings = "capability_mappings"
	// OptMarketMappings maps categories to market refs.
	OptMarketMappings = "market_mappings"
)

Well-known config option keys for cross-repo reference mappings. These map source system values to typed refs (e.g., "Acme Corp" → "customer:acme-001").

const (
	PriorityThin  = 0  // Native HTTP implementations
	PriorityThick = 10 // SDK-based implementations
)

Priority levels for provider registration. Higher priority providers override lower priority ones.

const (
	// MetaCurated indicates a pre-consolidated signal that should skip
	// the clustering stage of the consolidation pipeline. Use for signals
	// from sources that already aggregate user feedback (e.g., Aha Ideas
	// with votes and watchers). Value: bool.
	MetaCurated = "curated"
)

Well-known metadata keys for signal classification.

Variables

var (
	ErrNotSupported     = errors.New("operation not supported by this provider")
	ErrInvalidConfig    = errors.New("invalid provider configuration")
	ErrAuthentication   = errors.New("authentication failed")
	ErrRateLimited      = errors.New("rate limited by provider")
	ErrProviderNotFound = errors.New("provider not found in registry")
)

Common errors returned by providers.

Functions

func ClearRegistry

func ClearRegistry()

ClearRegistry removes all providers from the registry. Primarily useful for testing.

func GetPriority

func GetPriority(name string) int

GetPriority returns the priority of a registered provider. Returns -1 if the provider is not registered.

func IsCurated

func IsCurated(metadata map[string]any) bool

IsCurated checks if a signal is marked as curated (pre-consolidated). Curated signals should skip clustering and map directly to canonical signals.

func IsRegistered

func IsRegistered(name string) bool

IsRegistered checks if a provider is registered.

func List

func List() []string

List returns all registered provider names in alphabetical order.

func Register

func Register(name string, factory ProviderFactory, priority int)

Register adds a provider factory to the global registry.

If a provider with the same name exists, the one with higher priority wins. For equal priorities, the later registration wins.

Example:

func init() {
    omnisignal.Register("pagerduty", NewProvider, omnisignal.PriorityThick)
}

func Unregister

func Unregister(name string)

Unregister removes a provider from the registry. Primarily useful for testing.

func WrapErrorByMessage

func WrapErrorByMessage(err error, context string) error

WrapErrorByMessage wraps an error based on error message patterns when HTTP response is not available. Detects auth and rate-limit errors from common error message patterns.

func WrapHTTPError

func WrapHTTPError(err error, resp *http.Response, context string) error

WrapHTTPError wraps an error based on HTTP status code, returning a sentinel error (ErrAuthentication, ErrRateLimited) where appropriate. If resp is nil or the status doesn't match a sentinel, returns the original error wrapped with context.

Types

type Capabilities

type Capabilities struct {
	// SupportsStreaming indicates real-time signal streaming via Subscribe.
	SupportsStreaming bool

	// SupportsBatchFetch indicates efficient batch fetching.
	SupportsBatchFetch bool

	// SupportsFiltering indicates server-side filtering support.
	SupportsFiltering bool

	// SupportsAcknowledge indicates the ability to acknowledge signals.
	SupportsAcknowledge bool

	// MaxBatchSize is the maximum signals per fetch request.
	// Zero means no limit or unknown.
	MaxBatchSize int

	// RateLimitPerMinute is the provider's rate limit.
	// Zero means no limit or unknown.
	RateLimitPerMinute int

	// SignalTypes lists the signal types this provider can emit.
	SignalTypes []signal.Type
}

Capabilities describes what features a provider supports.

type Config

type Config struct {
	// APIKey is the primary authentication credential.
	APIKey string

	// APISecret is a secondary credential (if required).
	APISecret string

	// BaseURL overrides the default API endpoint.
	// Empty uses the provider's default.
	BaseURL string

	// Timeout for API requests. Zero uses provider default.
	Timeout time.Duration

	// RetryMax is the maximum number of retry attempts.
	// Zero means use provider default.
	RetryMax int

	// Options contains provider-specific configuration.
	Options map[string]any
}

Config holds provider configuration.

func (Config) GetOption

func (c Config) GetOption(key string, defaultVal any) any

GetOption retrieves a typed option value with a default fallback.

func (Config) GetStringMap

func (c Config) GetStringMap(key string) map[string]string

GetStringMap retrieves a map[string]string option. Handles both map[string]string and map[string]any with string values.

func (Config) GetStringOption

func (c Config) GetStringOption(key, defaultVal string) string

GetStringOption retrieves a string option with a default fallback.

type FetchOptions

type FetchOptions struct {
	// Since filters signals observed after this time (inclusive).
	Since time.Time

	// Until filters signals observed before this time (exclusive).
	// Zero value means no upper bound.
	Until time.Time

	// Limit is the maximum number of signals to return.
	// Zero means no limit (fetch all matching signals).
	Limit int

	// Statuses filters by signal status in the source system.
	// Empty means all statuses.
	Statuses []string

	// Severities filters by severity level.
	// Empty means all severities.
	Severities []string

	// Filters contains provider-specific filter parameters.
	// Keys and values are provider-dependent.
	Filters map[string]string
}

FetchOptions configures a fetch operation.

type Provider

type Provider interface {
	// Name returns the provider identifier (e.g., "pagerduty", "jira", "newrelic").
	Name() string

	// Fetch retrieves signals matching the given options.
	// Returns signals in chronological order (oldest first).
	// Implementations should handle pagination internally.
	Fetch(ctx context.Context, opts FetchOptions) ([]signal.Signal, error)

	// Subscribe opens a real-time stream of signals.
	// Returns ErrNotSupported if the provider doesn't support streaming.
	// The returned channel is closed when the context is canceled.
	Subscribe(ctx context.Context, opts SubscribeOptions) (<-chan signal.Signal, error)

	// Capabilities returns what this provider supports.
	Capabilities() Capabilities

	// Close releases any resources held by the provider.
	Close() error
}

Provider defines the interface for signal ingestion providers.

Implementations should be safe for concurrent use.

func MustNew

func MustNew(name string, cfg Config) Provider

MustNew creates a provider instance, panicking on error. Use only in initialization code where failure is unrecoverable.

func New

func New(name string, cfg Config) (Provider, error)

New creates a provider instance by name using the registered factory.

Returns ErrProviderNotFound if no provider is registered with that name.

type ProviderFactory

type ProviderFactory func(cfg Config) (Provider, error)

ProviderFactory creates a new provider instance from configuration.

type SubscribeOptions

type SubscribeOptions struct {
	// BufferSize is the channel buffer size for incoming signals.
	// Default is 100 if not specified.
	BufferSize int

	// Filters contains provider-specific filter parameters.
	Filters map[string]string
}

SubscribeOptions configures a subscription.