Automatically ingest trace topology (e.g., from OpenTelemetry) to build and update these DAGs dynamically

<html><head></head><body><h1>Feature Specification: Automatic Trace Topology Ingestion &amp; Dynamic DAG Building</h1>
<h2>1. Overview</h2>
<p>This feature enables NeuralBudget to automatically ingest distributed trace data from OpenTelemetry (and compatible observability systems) to dynamically construct and maintain Directed Acyclic Graphs (DAGs) representing service topologies and call flows. This eliminates manual service dependency definition and creates living, self-updating visualizations of system architecture.</p>
<h2>2. Problem Statement</h2>
<p>Current workflow limitations:</p>
<ul>
<li>DAG definitions are static and require manual updates when service architecture changes</li>
<li>Teams must manually maintain service dependency mappings</li>
<li>New services or changed communication patterns require code updates</li>
<li>No automatic detection of service relationships in distributed systems</li>
<li>Difficult to keep documentation in sync with actual runtime topology</li>
</ul>
<h2>3. Goals</h2>
<ul>
<li><strong>Automatic Discovery</strong>: Extract service topology directly from distributed trace telemetry</li>
<li><strong>Real-time Updates</strong>: Reflect architectural changes without manual intervention</li>
<li><strong>OpenTelemetry Native</strong>: Leverage OpenTelemetry standards for maximum compatibility</li>
<li><strong>Low Overhead</strong>: Minimal performance impact on trace processing</li>
<li><strong>Visualization</strong>: Auto-generate and update service dependency graphs</li>
<li><strong>Historical Tracking</strong>: Maintain topology versioning and change history</li>
</ul>
<h2>4. Scope</h2>
<h3>In Scope</h3>
<ul>
<li>OpenTelemetry trace ingestion (OTLP protocol)</li>
<li>Service and operation discovery from spans</li>
<li>Call relationship mapping (parent-child span relationships)</li>
<li>Dynamic DAG construction and updates</li>
<li>Topology versioning and change detection</li>
<li>REST APIs for topology queries</li>
<li>Basic visualization export (JSON, GraphML)</li>
</ul>
<h3>Out of Scope</h3>
<ul>
<li>Metrics/logs ingestion (trace-only in Phase 1)</li>
<li>Real-time visualization UI (API-first, UI integration separate)</li>
<li>Custom trace filtering rules (standard OpenTelemetry semantics only)</li>
<li>Cost optimization for trace sampling</li>
</ul>
<h2>5. Technical Architecture</h2>
<h3>5.1 Core Components</h3>
<pre><code>┌─────────────────────────────────────────────────────────┐
│                  OpenTelemetry Collector                │
│                   (OTLP Receiver)                       │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│            Trace Ingestion Service                       │
│  - Validate &amp; normalize traces                          │
│  - Extract span metadata                                │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│         Topology Discovery Engine                        │
│  - Build service graph from spans                       │
│  - Detect operation flows                               │
│  - Identify communication patterns                      │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│              DAG Builder &amp; Manager                       │
│  - Construct/update DAGs                                │
│  - Detect topology changes                              │
│  - Version management                                   │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│          Topology Store (Database)                       │
│  - Current topology state                               │
│  - Historical versions                                  │
│  - Change audit log                                     │
└─────────────────────────────────────────────────────────┘
</code></pre>
<h3>5.2 Data Flow</h3>
<ol>
<li><strong>Trace Reception</strong>: OpenTelemetry traces arrive via OTLP (gRPC/HTTP)</li>
<li><strong>Normalization</strong>: Extract key attributes (service.name, span.kind, attributes)</li>
<li><strong>Relationship Detection</strong>: Identify parent-child relationships and RPC calls</li>
<li><strong>Service Mapping</strong>: Group spans by service and build adjacency matrix</li>
<li><strong>DAG Construction</strong>: Create graph nodes (services) and edges (calls)</li>
<li><strong>Change Detection</strong>: Compare against previous topology state</li>
<li><strong>Storage &amp; Versioning</strong>: Persist changes with timestamps and metadata</li>
<li><strong>Notification</strong>: Emit events for topology changes</li>
</ol>
<h2>6. Data Model</h2>
<h3>6.1 Key Span Attributes (OpenTelemetry Semantic Conventions)</h3>
<pre><code class="language-json">{
  "service.name": "payment-service",
  "service.version": "1.2.3",
  "span.kind": "CLIENT|SERVER|PRODUCER|CONSUMER|INTERNAL",
  "span.name": "PaymentService.processPayment",
  "rpc.service": "payment.v1.PaymentService",
  "rpc.method": "ProcessPayment",
  "http.method": "POST",
  "http.target": "/api/v1/payments",
  "peer.service": "database-service",
  "messaging.system": "kafka",
  "messaging.destination": "payment-events"
}
</code></pre>
<h3>6.2 Service Node</h3>
<pre><code class="language-typescript">interface ServiceNode {
  id: string;                          // Unique service identifier
  name: string;                        // service.name
  version: string;                     // service.version (nullable)
  kind: 'SERVICE' | 'DATABASE' | 'QUEUE' | 'EXTERNAL';
  firstSeen: ISO8601DateTime;
  lastSeen: ISO8601DateTime;
  metadata: {
    environment?: string;
    region?: string;
    language?: string;
    [key: string]: any;
  };
  operations: string[];               // List of span.name values
}
</code></pre>
<h3>6.3 Edge (Call Relationship)</h3>
<pre><code class="language-typescript">interface ServiceEdge {
  id: string;                          // Unique edge identifier
  sourceId: string;                    // caller service
  targetId: string;                    // callee service
  protocol: 'HTTP' | 'GRPC' | 'AMQP' | 'KAFKA' | 'DATABASE' | 'UNKNOWN';
  callCount: number;                   // Cumulative call count
  errorCount: number;                  // Failed calls in window
  latencyP50: number;                  // ms
  latencyP99: number;                  // ms
  firstSeen: ISO8601DateTime;
  lastSeen: ISO8601DateTime;
  operations: string[];                // RPC/HTTP operations
  attributes: {
    [key: string]: any;
  };
}
</code></pre>
<h3>6.4 DAG Version</h3>
<pre><code class="language-typescript">interface DAGVersion {
  id: string;                          // UUID
  version: number;                     // Incremental counter
  timestamp: ISO8601DateTime;
  nodes: ServiceNode[];
  edges: ServiceEdge[];
  changeMetadata: {
    nodesAdded: string[];              // Node IDs
    nodesRemoved: string[];
    edgesAdded: string[];
    edgesRemoved: string[];
    nodesModified: string[];           // version/metadata changes
    reason: string;                    // "topology_discovery" | "trace_analysis"
  };
  hash: string;                        // SHA256 of topology
}
</code></pre>
<h2>7. Functional Requirements</h2>
<h3>FR1: Trace Ingestion</h3>
<ul>
<li>Accept OpenTelemetry traces via OTLP (gRPC + HTTP)</li>
<li>Validate trace format and schema</li>
<li>Handle batch and streaming ingestion</li>
<li>Support configurable retention window (default: 7 days)</li>
</ul>
<h3>FR2: Service Discovery</h3>
<ul>
<li>Extract service.name from trace attributes</li>
<li>Identify service kind (service vs external dependency)</li>
<li>Track service versions when available</li>
<li>Detect first/last seen timestamps</li>
</ul>
<h3>FR3: Call Relationship Detection</h3>
<ul>
<li>Identify parent-child span relationships</li>
<li>Classify communication protocol (HTTP, gRPC, DB, Messaging)</li>
<li>Extract operation names and endpoints</li>
<li>Calculate call frequency and error rates</li>
<li>Measure latency percentiles (p50, p99)</li>
</ul>
<h3>FR4: DAG Construction</h3>
<ul>
<li>Build service graph from discovered relationships</li>
<li>Ensure DAG remains acyclic (cycle detection)</li>
<li>Merge redundant relationships</li>
<li>Support multiple versions of DAG (historical)</li>
</ul>
<h3>FR5: Change Detection &amp; Notification</h3>
<ul>
<li>Compare current topology against previous state</li>
<li>Classify changes (additions, removals, modifications)</li>
<li>Generate audit trail with reasons</li>
<li>Emit events for topology changes (webhook, queue)</li>
</ul>
<h3>FR6: Query &amp; Export APIs</h3>
<ul>
<li>List all services in current topology</li>
<li>Get service details with operations</li>
<li>Query edges between services</li>
<li>Export DAG in multiple formats (JSON, GraphML, DOT)</li>
<li>Historical topology queries</li>
<li>Change log queries</li>
</ul>
<h3>FR7: Aggregation Window</h3>
<ul>
<li>Configure time window for topology analysis (default: 1 hour)</li>
<li>Aggregate metrics within window (call counts, latencies)</li>
<li>Generate new DAG version periodically or on-demand</li>
</ul>
<h2>8. Non-Functional Requirements</h2>
<h3>NFR1: Performance</h3>
<ul>
<li>Ingest 100k+ spans/second with &lt;100ms latency</li>
<li>Topology update &lt;5 seconds for most changes</li>
<li>Query API response time &lt;200ms (p99)</li>
</ul>
<h3>NFR2: Reliability</h3>
<ul>
<li>99.9% availability for ingestion pipeline</li>
<li>Graceful degradation (buffer overflow handling)</li>
<li>Automatic retry for failed trace processing</li>
</ul>
<h3>NFR3: Scalability</h3>
<ul>
<li>Horizontal scaling of trace processors</li>
<li>Database sharding support for large deployments</li>
<li>Efficient memory usage for large topologies (1000+ services)</li>
</ul>
<h3>NFR4: Data Quality</h3>
<ul>
<li>Deduplicate spans with matching IDs</li>
<li>Validate semantic convention compliance</li>
<li>Configurable leniency for malformed traces</li>
</ul>
<h3>NFR5: Storage</h3>
<ul>
<li>Compress historical DAG versions</li>
<li>Configurable TTL for version history (default: 30 days)</li>
<li>Efficient storage &lt;10MB per DAG snapshot</li>
</ul>
<h2>9. API Specification</h2>
<h3>9.1 Trace Ingestion Endpoint</h3>
<pre><code>POST /v1/traces
Content-Type: application/protobuf

Request Body: ExportTraceServiceRequest (OTLP format)
Response: ExportTraceServiceResponse (202 Accepted)
</code></pre>
<h3>9.2 Topology Query APIs</h3>
<pre><code># Get current service topology
GET /v1/topology/services
Response:
{
  "services": [ServiceNode],
  "timestamp": "2024-01-15T10:30:00Z",
  "version": 42
}

# Get service details
GET /v1/topology/services/{serviceId}
Response: ServiceNode + operations array + edge statistics

# Get call relationships
GET /v1/topology/edges?source={serviceId}&amp;target={targetId}
Response: {
  "edges": [ServiceEdge],
  "aggregationWindow": "1h"
}

# Get DAG for visualization
GET /v1/topology/dag?format={json|graphml|dot}
Response: 
- JSON: { "nodes": [...], "edges": [...] }
- GraphML/DOT: Graph format

# List DAG versions
GET /v1/topology/versions?limit=10&amp;offset=0
Response: {
  "versions": [DAGVersion],
  "total": 150
}

# Get specific DAG version
GET /v1/topology/versions/{versionId}
Response: DAGVersion

# Get topology changes
GET /v1/topology/changes?since={timestamp}&amp;limit=50
Response: {
  "changes": [ChangeMetadata],
  "hasMore": boolean
}
</code></pre>
<h3>9.3 Configuration API</h3>
<pre><code># Update topology discovery settings
POST /v1/config/topology
{
  "aggregationWindow": "1h",
  "retentionDays": 30,
  "cyclicityCheckEnabled": true,
  "minCallThreshold": 1,
  "externalServicePatterns": ["*.external.com"]
}

# Define service kind overrides
POST /v1/config/services/{serviceName}
{
  "kind": "DATABASE",
  "metadata": {
    "environment": "production",
    "team": "platform"
  }
}
</code></pre>
<h2>10. Implementation Details</h2>
<h3>10.1 Technology Stack</h3>
<ul>
<li><strong>Language</strong>: (Match NeuralBudget's existing stack)</li>
<li><strong>Trace Processing</strong>: OpenTelemetry SDK / Collector</li>
<li><strong>Graph Library</strong>: networkx (Python) or igraph</li>
<li><strong>Storage</strong>: PostgreSQL with JSONB (topologies) + TimescaleDB (metrics)</li>
<li><strong>Streaming</strong>: Optional: Kafka for high-volume scenarios</li>
<li><strong>Change Events</strong>: Pub/Sub or Webhooks</li>
</ul>
<h3>10.2 Topology Discovery Algorithm</h3>
<pre><code>Algorithm: DiscoverTopology(traces, aggregationWindow)

Input: List of traces from past aggregationWindow
Output: DAG with services (nodes) and calls (edges)

1. groupByService(traces) → traces_by_service
2. for each service_trace in traces_by_service:
     - extract service_name, version, metadata
     - create ServiceNode if not exists
     - update lastSeen, version, metadata
   
3. for each trace in traces:
     for each span in trace:
       - if span.kind == SERVER or CONSUMER:
         - extract peer.service or rpc.service
         - if parentSpan exists and parentSpan.service != current.service:
           - create Edge(parentSpan.service → span.service)
           - update protocol, operation, latency metrics

4. deduplicateEdges() → merge parallel edges
5. validateDAG() → check acyclicity
6. compareWithPreviousVersion() → detect changes
7. persistDAG(version)
8. emitChangeEvents()

Time Complexity: O(n) where n = number of spans
Space Complexity: O(s + e) where s = services, e = edges
</code></pre>
<h3>10.3 Cycle Detection &amp; Resolution</h3>
<pre><code>Algorithm: DetectAndResolveCycles()

1. Use DFS to identify cycles in current DAG
2. If cycles detected:
   - Log warning with cycle details
   - Option A: Remove lowest-confidence edges (weak signals)
   - Option B: Mark services as "potential cycle" with metadata
   - Option C: Flag for manual review
3. Preserve topology for operational continuity
</code></pre>
<h3>10.4 Change Detection Algorithm</h3>
<pre><code>Algorithm: DetectChanges(currentDAG, previousDAG)

Input: Two DAG versions
Output: Change metadata

1. nodeChanges = diff(currentDAG.nodes, previousDAG.nodes)
   - nodesAdded: nodes in current but not previous
   - nodesRemoved: nodes in previous but not current
   - nodesModified: version or metadata changes

2. edgeChanges = diff(currentDAG.edges, previousDAG.edges)
   - edgesAdded: new communication patterns
   - edgesRemoved: deprecated communication paths
   - edgesModified: protocol/latency threshold changes

3. generateChangeNotification()
</code></pre>
<h2>11. Configuration</h2>
<h3>11.1 Environment Variables</h3>
<pre><code class="language-bash"># Trace ingestion
TRACE_OTLP_PORT=4317                    # gRPC endpoint
TRACE_HTTP_PORT=4318                    # HTTP endpoint
TRACE_BATCH_SIZE=1000
TRACE_BATCH_TIMEOUT_MS=5000

# Topology discovery
TOPOLOGY_AGGREGATION_WINDOW=3600        # seconds (1 hour)
TOPOLOGY_DISCOVERY_INTERVAL=600         # seconds (10 minutes)
TOPOLOGY_RETENTION_DAYS=30
TOPOLOGY_VERSION_LIMIT=500

# Change detection
TOPOLOGY_CHANGE_THRESHOLD=0.05          # 5% change triggers new version
TOPOLOGY_CHANGE_WEBHOOK_URL=            # optional webhook
TOPOLOGY_CHANGE_EVENTS_ENABLED=true

# Storage
POSTGRES_CONNECTION_STRING=
TIMESCALEDB_ENABLED=true
METRICS_WINDOW_SIZE=3600                # seconds

# Performance
TRACE_PROCESSOR_WORKERS=4
DAG_BUILDER_TIMEOUT_MS=5000
CYCLE_CHECK_ENABLED=true
</code></pre>
<h3>11.2 Configuration File Example</h3>
<pre><code class="language-yaml"># config.yaml
tracing:
  otlp:
    enabled: true
    port: 4317
  batch:
    size: 1000
    timeout_ms: 5000

topology:
  discovery:
    enabled: true
    aggregation_window: 1h
    interval: 10m
    min_call_threshold: 1
  
  storage:
    retention: 30d
    max_versions: 500
    compression: gzip
  
  validation:
    enable_cycle_check: true
    external_service_patterns:
      - "*.amazonaws.com"
      - "*.stripe.com"
      - "*.external.io"
  
  change_detection:
    enabled: true
    threshold: 0.05  # 5%
    notifications:
      webhook_url: "https://events.company.com/topology-changes"
      queue_topic: "topology.changes"
</code></pre>
<h2>12. Example Usage</h2>
<h3>12.1 Sending Traces</h3>
<pre><code class="language-python">from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Configure OTLP exporter pointing to NeuralBudget
otlp_exporter = OTLPSpanExporter(
    endpoint="localhost:4317"
)

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(otlp_exporter)
)

tracer = trace.get_tracer(__name__)

# Spans automatically contribute to topology discovery
with tracer.start_as_current_span("PaymentService.processPayment") as span:
    span.set_attribute("service.name", "payment-service")
    # ... business logic
</code></pre>
<h3>12.2 Querying Topology</h3>
<pre><code class="language-bash"># Get current service topology
curl http://localhost:8080/v1/topology/services | jq

# Export as GraphML for visualization
curl http://localhost:8080/v1/topology/dag?format=graphml &gt; topology.graphml

# Check what changed in last hour
curl "http://localhost:8080/v1/topology/changes?since=2024-01-15T09:00:00Z" | jq

# Get service dependencies
curl http://localhost:8080/v1/topology/edges?source=payment-service | jq
</code></pre>
<h2>13. Error Handling</h2>
<h3>13.1 Trace Processing Errors</h3>

Error | Handling
-- | --
Malformed trace | Log warning, skip span, increment error counter
Missing service.name | Use span.peer_service or rpc.service, fallback to "unknown-service"
Inconsistent versions | Use most recent version seen
Dropped due to backpressure | Emit metric, return HTTP 429


<h2>14. Testing Strategy</h2>
<h3>14.1 Unit Tests</h3>
<ul>
<li>Span attribute extraction and validation</li>
<li>Service node creation and deduplication</li>
<li>Edge relationship building</li>
<li>Cycle detection algorithm</li>
<li>Change detection logic</li>
</ul>
<h3>14.2 Integration Tests</h3>
<ul>
<li>Full trace → topology pipeline</li>
<li>Multiple concurrent services</li>
<li>Protocol detection (HTTP, gRPC, DB, Messaging)</li>
<li>Version management and history</li>
</ul>
<h3>14.3 Load Testing</h3>
<ul>
<li>100k spans/second ingestion</li>
<li>10,000 service DAG construction</li>
<li>Query response times under load</li>
<li>Memory/storage efficiency</li>
</ul>
<h3>14.4 Test Data</h3>
<pre><code class="language-python"># Generate synthetic traces
from opentelemetry.proto.trace.v1 import traces_pb2

def generate_synthetic_trace(
    num_spans=100,
    num_services=10,
    error_rate=0.05
):
    # Creates realistic trace with multiple services
    # for testing topology discovery
    pass
</code></pre>
<h2>15. Monitoring &amp; Observability</h2>
<h3>15.1 Key Metrics</h3>
<pre><code># Trace processing
trace_ingestion_rate          (spans/sec)
trace_processing_latency      (ms, p50/p99)
trace_processing_errors       (count)
trace_buffer_capacity         (%)

# Topology discovery
services_discovered           (count)
service_edges_detected        (count)
topology_update_latency       (ms)
topology_version_changes      (count)
cycles_detected               (count)

# Storage
topology_storage_size         (bytes)
version_history_count         (count)
database_query_latency        (ms)
</code></pre>
<h3>15.2 Alerting Rules</h3>
<pre><code class="language-yaml">alerts:
  - name: HighTraceProcessingError
    condition: trace_processing_errors &gt; 100/min
    severity: warning
    
  - name: TopologyDiscoveryFailure
    condition: topology_update_latency &gt; 30s
    severity: critical
    
  - name: CycleDetected
    condition: cycles_detected &gt; 0
    severity: warning
</code></pre>
<h2>16. Future Enhancements</h2>
<ul>
<li><strong>Phase 2</strong>: Metrics &amp; logs correlation with traces</li>
<li><strong>Phase 3</strong>: ML-based anomaly detection in topology changes</li>
<li><strong>Phase 4</strong>: Service mesh integration (Istio, Linkerd)</li>
<li><strong>Phase 5</strong>: Real-time Web UI with live topology updates</li>
<li><strong>Phase 6</strong>: Trace-based SLA tracking and compliance</li>
<li><strong>Phase 7</strong>: Cost attribution via topology paths</li>
</ul>
<h2>17. Migration &amp; Rollout</h2>
<h3>17.1 Backward Compatibility</h3>
<ul>
<li>DAG APIs versioned (<code>/v1/</code>, <code>/v2/</code>)</li>
<li>Manual DAG definition still supported</li>
<li>Gradual migration of users to auto-discovery</li>
</ul>
<h3>17.2 Rollout Plan</h3>
<ol>
<li><strong>Alpha</strong>: Internal testing with one production service</li>
<li><strong>Beta</strong>: Gradual rollout to 10% of users, with manual DAG override option</li>
<li><strong>GA</strong>: Full rollout, auto-discovery as default</li>
<li><strong>Deprecation</strong>: Manual DAG definition deprecated in 12 months</li>
</ol>
<h2>18. Success Metrics</h2>
<ul>
<li>Topology accuracy &gt;99% vs. manual definitions</li>
<li>95% reduction in manual DAG updates</li>
<li>&lt;5 second detection latency for new services</li>
<li>&lt;2% false positive change detection rate</li>
<li>99.9% availability of topology APIs</li>
<li>Ingestion throughput &gt;100k spans/sec</li>
</ul>
<h2>19. References</h2>
<ul>
<li><a href="https://opentelemetry.io/docs/specs/otel/">OpenTelemetry Specification</a></li>
<li><a href="https://opentelemetry.io/docs/specs/semconv/">OpenTelemetry Semantic Conventions</a></li>
<li><a href="https://github.com/open-telemetry/opentelemetry-proto">OTLP Protocol</a></li>
<li><a href="https://en.wikipedia.org/wiki/Cycle_(graph_theory)">Graph Theory - Cycle Detection</a></li>
</ul></body></html># Feature Specification: Automatic Trace Topology Ingestion & Dynamic DAG Building

## 1. Overview

This feature enables NeuralBudget to automatically ingest distributed trace data from OpenTelemetry (and compatible observability systems) to dynamically construct and maintain Directed Acyclic Graphs (DAGs) representing service topologies and call flows. This eliminates manual service dependency definition and creates living, self-updating visualizations of system architecture.

## 2. Problem Statement

Current workflow limitations:
- DAG definitions are static and require manual updates when service architecture changes
- Teams must manually maintain service dependency mappings
- New services or changed communication patterns require code updates
- No automatic detection of service relationships in distributed systems
- Difficult to keep documentation in sync with actual runtime topology

## 3. Goals

- **Automatic Discovery**: Extract service topology directly from distributed trace telemetry
- **Real-time Updates**: Reflect architectural changes without manual intervention
- **OpenTelemetry Native**: Leverage OpenTelemetry standards for maximum compatibility
- **Low Overhead**: Minimal performance impact on trace processing
- **Visualization**: Auto-generate and update service dependency graphs
- **Historical Tracking**: Maintain topology versioning and change history

## 4. Scope

### In Scope
- OpenTelemetry trace ingestion (OTLP protocol)
- Service and operation discovery from spans
- Call relationship mapping (parent-child span relationships)
- Dynamic DAG construction and updates
- Topology versioning and change detection
- REST APIs for topology queries
- Basic visualization export (JSON, GraphML)

### Out of Scope
- Metrics/logs ingestion (trace-only in Phase 1)
- Real-time visualization UI (API-first, UI integration separate)
- Custom trace filtering rules (standard OpenTelemetry semantics only)
- Cost optimization for trace sampling

## 5. Technical Architecture

### 5.1 Core Components

```
┌─────────────────────────────────────────────────────────┐
│                  OpenTelemetry Collector                │
│                   (OTLP Receiver)                       │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│            Trace Ingestion Service                       │
│  - Validate & normalize traces                          │
│  - Extract span metadata                                │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│         Topology Discovery Engine                        │
│  - Build service graph from spans                       │
│  - Detect operation flows                               │
│  - Identify communication patterns                      │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│              DAG Builder & Manager                       │
│  - Construct/update DAGs                                │
│  - Detect topology changes                              │
│  - Version management                                   │
└──────────────────────┬──────────────────────────────────┘
                       │
                       ▼
┌─────────────────────────────────────────────────────────┐
│          Topology Store (Database)                       │
│  - Current topology state                               │
│  - Historical versions                                  │
│  - Change audit log                                     │
└─────────────────────────────────────────────────────────┘
```

### 5.2 Data Flow

1. **Trace Reception**: OpenTelemetry traces arrive via OTLP (gRPC/HTTP)
2. **Normalization**: Extract key attributes (service.name, span.kind, attributes)
3. **Relationship Detection**: Identify parent-child relationships and RPC calls
4. **Service Mapping**: Group spans by service and build adjacency matrix
5. **DAG Construction**: Create graph nodes (services) and edges (calls)
6. **Change Detection**: Compare against previous topology state
7. **Storage & Versioning**: Persist changes with timestamps and metadata
8. **Notification**: Emit events for topology changes

## 6. Data Model

### 6.1 Key Span Attributes (OpenTelemetry Semantic Conventions)

```json
{
  "service.name": "payment-service",
  "service.version": "1.2.3",
  "span.kind": "CLIENT|SERVER|PRODUCER|CONSUMER|INTERNAL",
  "span.name": "PaymentService.processPayment",
  "rpc.service": "payment.v1.PaymentService",
  "rpc.method": "ProcessPayment",
  "http.method": "POST",
  "http.target": "/api/v1/payments",
  "peer.service": "database-service",
  "messaging.system": "kafka",
  "messaging.destination": "payment-events"
}
```

### 6.2 Service Node

```typescript
interface ServiceNode {
  id: string;                          // Unique service identifier
  name: string;                        // service.name
  version: string;                     // service.version (nullable)
  kind: 'SERVICE' | 'DATABASE' | 'QUEUE' | 'EXTERNAL';
  firstSeen: ISO8601DateTime;
  lastSeen: ISO8601DateTime;
  metadata: {
    environment?: string;
    region?: string;
    language?: string;
    [key: string]: any;
  };
  operations: string[];               // List of span.name values
}
```

### 6.3 Edge (Call Relationship)

```typescript
interface ServiceEdge {
  id: string;                          // Unique edge identifier
  sourceId: string;                    // caller service
  targetId: string;                    // callee service
  protocol: 'HTTP' | 'GRPC' | 'AMQP' | 'KAFKA' | 'DATABASE' | 'UNKNOWN';
  callCount: number;                   // Cumulative call count
  errorCount: number;                  // Failed calls in window
  latencyP50: number;                  // ms
  latencyP99: number;                  // ms
  firstSeen: ISO8601DateTime;
  lastSeen: ISO8601DateTime;
  operations: string[];                // RPC/HTTP operations
  attributes: {
    [key: string]: any;
  };
}
```

### 6.4 DAG Version

```typescript
interface DAGVersion {
  id: string;                          // UUID
  version: number;                     // Incremental counter
  timestamp: ISO8601DateTime;
  nodes: ServiceNode[];
  edges: ServiceEdge[];
  changeMetadata: {
    nodesAdded: string[];              // Node IDs
    nodesRemoved: string[];
    edgesAdded: string[];
    edgesRemoved: string[];
    nodesModified: string[];           // version/metadata changes
    reason: string;                    // "topology_discovery" | "trace_analysis"
  };
  hash: string;                        // SHA256 of topology
}
```

## 7. Functional Requirements

### FR1: Trace Ingestion
- Accept OpenTelemetry traces via OTLP (gRPC + HTTP)
- Validate trace format and schema
- Handle batch and streaming ingestion
- Support configurable retention window (default: 7 days)

### FR2: Service Discovery
- Extract service.name from trace attributes
- Identify service kind (service vs external dependency)
- Track service versions when available
- Detect first/last seen timestamps

### FR3: Call Relationship Detection
- Identify parent-child span relationships
- Classify communication protocol (HTTP, gRPC, DB, Messaging)
- Extract operation names and endpoints
- Calculate call frequency and error rates
- Measure latency percentiles (p50, p99)

### FR4: DAG Construction
- Build service graph from discovered relationships
- Ensure DAG remains acyclic (cycle detection)
- Merge redundant relationships
- Support multiple versions of DAG (historical)

### FR5: Change Detection & Notification
- Compare current topology against previous state
- Classify changes (additions, removals, modifications)
- Generate audit trail with reasons
- Emit events for topology changes (webhook, queue)

### FR6: Query & Export APIs
- List all services in current topology
- Get service details with operations
- Query edges between services
- Export DAG in multiple formats (JSON, GraphML, DOT)
- Historical topology queries
- Change log queries

### FR7: Aggregation Window
- Configure time window for topology analysis (default: 1 hour)
- Aggregate metrics within window (call counts, latencies)
- Generate new DAG version periodically or on-demand

## 8. Non-Functional Requirements

### NFR1: Performance
- Ingest 100k+ spans/second with <100ms latency
- Topology update <5 seconds for most changes
- Query API response time <200ms (p99)

### NFR2: Reliability
- 99.9% availability for ingestion pipeline
- Graceful degradation (buffer overflow handling)
- Automatic retry for failed trace processing

### NFR3: Scalability
- Horizontal scaling of trace processors
- Database sharding support for large deployments
- Efficient memory usage for large topologies (1000+ services)

### NFR4: Data Quality
- Deduplicate spans with matching IDs
- Validate semantic convention compliance
- Configurable leniency for malformed traces

### NFR5: Storage
- Compress historical DAG versions
- Configurable TTL for version history (default: 30 days)
- Efficient storage <10MB per DAG snapshot

## 9. API Specification

### 9.1 Trace Ingestion Endpoint

```
POST /v1/traces
Content-Type: application/protobuf

Request Body: ExportTraceServiceRequest (OTLP format)
Response: ExportTraceServiceResponse (202 Accepted)
```

### 9.2 Topology Query APIs

```
# Get current service topology
GET /v1/topology/services
Response:
{
  "services": [ServiceNode],
  "timestamp": "2024-01-15T10:30:00Z",
  "version": 42
}

# Get service details
GET /v1/topology/services/{serviceId}
Response: ServiceNode + operations array + edge statistics

# Get call relationships
GET /v1/topology/edges?source={serviceId}&target={targetId}
Response: {
  "edges": [ServiceEdge],
  "aggregationWindow": "1h"
}

# Get DAG for visualization
GET /v1/topology/dag?format={json|graphml|dot}
Response: 
- JSON: { "nodes": [...], "edges": [...] }
- GraphML/DOT: Graph format

# List DAG versions
GET /v1/topology/versions?limit=10&offset=0
Response: {
  "versions": [DAGVersion],
  "total": 150
}

# Get specific DAG version
GET /v1/topology/versions/{versionId}
Response: DAGVersion

# Get topology changes
GET /v1/topology/changes?since={timestamp}&limit=50
Response: {
  "changes": [ChangeMetadata],
  "hasMore": boolean
}
```

### 9.3 Configuration API

```
# Update topology discovery settings
POST /v1/config/topology
{
  "aggregationWindow": "1h",
  "retentionDays": 30,
  "cyclicityCheckEnabled": true,
  "minCallThreshold": 1,
  "externalServicePatterns": ["*.external.com"]
}

# Define service kind overrides
POST /v1/config/services/{serviceName}
{
  "kind": "DATABASE",
  "metadata": {
    "environment": "production",
    "team": "platform"
  }
}
```

## 10. Implementation Details

### 10.1 Technology Stack

- **Language**: (Match NeuralBudget's existing stack)
- **Trace Processing**: OpenTelemetry SDK / Collector
- **Graph Library**: networkx (Python) or igraph
- **Storage**: PostgreSQL with JSONB (topologies) + TimescaleDB (metrics)
- **Streaming**: Optional: Kafka for high-volume scenarios
- **Change Events**: Pub/Sub or Webhooks

### 10.2 Topology Discovery Algorithm

```
Algorithm: DiscoverTopology(traces, aggregationWindow)

Input: List of traces from past aggregationWindow
Output: DAG with services (nodes) and calls (edges)

1. groupByService(traces) → traces_by_service
2. for each service_trace in traces_by_service:
     - extract service_name, version, metadata
     - create ServiceNode if not exists
     - update lastSeen, version, metadata
   
3. for each trace in traces:
     for each span in trace:
       - if span.kind == SERVER or CONSUMER:
         - extract peer.service or rpc.service
         - if parentSpan exists and parentSpan.service != current.service:
           - create Edge(parentSpan.service → span.service)
           - update protocol, operation, latency metrics

4. deduplicateEdges() → merge parallel edges
5. validateDAG() → check acyclicity
6. compareWithPreviousVersion() → detect changes
7. persistDAG(version)
8. emitChangeEvents()

Time Complexity: O(n) where n = number of spans
Space Complexity: O(s + e) where s = services, e = edges
```

### 10.3 Cycle Detection & Resolution

```
Algorithm: DetectAndResolveCycles()

1. Use DFS to identify cycles in current DAG
2. If cycles detected:
   - Log warning with cycle details
   - Option A: Remove lowest-confidence edges (weak signals)
   - Option B: Mark services as "potential cycle" with metadata
   - Option C: Flag for manual review
3. Preserve topology for operational continuity
```

### 10.4 Change Detection Algorithm

```
Algorithm: DetectChanges(currentDAG, previousDAG)

Input: Two DAG versions
Output: Change metadata

1. nodeChanges = diff(currentDAG.nodes, previousDAG.nodes)
   - nodesAdded: nodes in current but not previous
   - nodesRemoved: nodes in previous but not current
   - nodesModified: version or metadata changes

2. edgeChanges = diff(currentDAG.edges, previousDAG.edges)
   - edgesAdded: new communication patterns
   - edgesRemoved: deprecated communication paths
   - edgesModified: protocol/latency threshold changes

3. generateChangeNotification()
```

## 11. Configuration

### 11.1 Environment Variables

```bash
# Trace ingestion
TRACE_OTLP_PORT=4317                    # gRPC endpoint
TRACE_HTTP_PORT=4318                    # HTTP endpoint
TRACE_BATCH_SIZE=1000
TRACE_BATCH_TIMEOUT_MS=5000

# Topology discovery
TOPOLOGY_AGGREGATION_WINDOW=3600        # seconds (1 hour)
TOPOLOGY_DISCOVERY_INTERVAL=600         # seconds (10 minutes)
TOPOLOGY_RETENTION_DAYS=30
TOPOLOGY_VERSION_LIMIT=500

# Change detection
TOPOLOGY_CHANGE_THRESHOLD=0.05          # 5% change triggers new version
TOPOLOGY_CHANGE_WEBHOOK_URL=            # optional webhook
TOPOLOGY_CHANGE_EVENTS_ENABLED=true

# Storage
POSTGRES_CONNECTION_STRING=
TIMESCALEDB_ENABLED=true
METRICS_WINDOW_SIZE=3600                # seconds

# Performance
TRACE_PROCESSOR_WORKERS=4
DAG_BUILDER_TIMEOUT_MS=5000
CYCLE_CHECK_ENABLED=true
```

### 11.2 Configuration File Example

```yaml
# config.yaml
tracing:
  otlp:
    enabled: true
    port: 4317
  batch:
    size: 1000
    timeout_ms: 5000

topology:
  discovery:
    enabled: true
    aggregation_window: 1h
    interval: 10m
    min_call_threshold: 1
  
  storage:
    retention: 30d
    max_versions: 500
    compression: gzip
  
  validation:
    enable_cycle_check: true
    external_service_patterns:
      - "*.amazonaws.com"
      - "*.stripe.com"
      - "*.external.io"
  
  change_detection:
    enabled: true
    threshold: 0.05  # 5%
    notifications:
      webhook_url: "https://events.company.com/topology-changes"
      queue_topic: "topology.changes"
```

## 12. Example Usage

### 12.1 Sending Traces

```python
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Configure OTLP exporter pointing to NeuralBudget
otlp_exporter = OTLPSpanExporter(
    endpoint="localhost:4317"
)

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(otlp_exporter)
)

tracer = trace.get_tracer(__name__)

# Spans automatically contribute to topology discovery
with tracer.start_as_current_span("PaymentService.processPayment") as span:
    span.set_attribute("service.name", "payment-service")
    # ... business logic
```

### 12.2 Querying Topology

```bash
# Get current service topology
curl http://localhost:8080/v1/topology/services | jq

# Export as GraphML for visualization
curl http://localhost:8080/v1/topology/dag?format=graphml > topology.graphml

# Check what changed in last hour
curl "http://localhost:8080/v1/topology/changes?since=2024-01-15T09:00:00Z" | jq

# Get service dependencies
curl http://localhost:8080/v1/topology/edges?source=payment-service | jq
```

## 13. Error Handling

### 13.1 Trace Processing Errors

| Error | Handling |
|-------|----------|
| Malformed trace | Log warning, skip span, increment error counter |
| Missing service.name | Use span.peer_service or rpc.service, fallback to "unknown-service" |
| Inconsistent versions | Use most recent version seen |
| Dropped due to backpressure | Emit metric, return HTTP 429 |

### 13.2 DAG Construction Errors

| Error | Handling |
|-------|----------|
| Detected cycle | Log warning, remove weakest edge, flag for review |
| Topology explosion (>5000 nodes) | Aggregate by domain, emit alert |
| Failed change detection | Skip version, retry on next interval |

## 14. Testing Strategy

### 14.1 Unit Tests
- Span attribute extraction and validation
- Service node creation and deduplication
- Edge relationship building
- Cycle detection algorithm
- Change detection logic

### 14.2 Integration Tests
- Full trace → topology pipeline
- Multiple concurrent services
- Protocol detection (HTTP, gRPC, DB, Messaging)
- Version management and history

### 14.3 Load Testing
- 100k spans/second ingestion
- 10,000 service DAG construction
- Query response times under load
- Memory/storage efficiency

### 14.4 Test Data
```python
# Generate synthetic traces
from opentelemetry.proto.trace.v1 import traces_pb2

def generate_synthetic_trace(
    num_spans=100,
    num_services=10,
    error_rate=0.05
):
    # Creates realistic trace with multiple services
    # for testing topology discovery
    pass
```

## 15. Monitoring & Observability

### 15.1 Key Metrics

```
# Trace processing
trace_ingestion_rate          (spans/sec)
trace_processing_latency      (ms, p50/p99)
trace_processing_errors       (count)
trace_buffer_capacity         (%)

# Topology discovery
services_discovered           (count)
service_edges_detected        (count)
topology_update_latency       (ms)
topology_version_changes      (count)
cycles_detected               (count)

# Storage
topology_storage_size         (bytes)
version_history_count         (count)
database_query_latency        (ms)
```

### 15.2 Alerting Rules

```yaml
alerts:
  - name: HighTraceProcessingError
    condition: trace_processing_errors > 100/min
    severity: warning
    
  - name: TopologyDiscoveryFailure
    condition: topology_update_latency > 30s
    severity: critical
    
  - name: CycleDetected
    condition: cycles_detected > 0
    severity: warning
```

## 16. Future Enhancements

- **Phase 2**: Metrics & logs correlation with traces
- **Phase 3**: ML-based anomaly detection in topology changes
- **Phase 4**: Service mesh integration (Istio, Linkerd)
- **Phase 5**: Real-time Web UI with live topology updates
- **Phase 6**: Trace-based SLA tracking and compliance
- **Phase 7**: Cost attribution via topology paths

## 17. Migration & Rollout

### 17.1 Backward Compatibility
- DAG APIs versioned (`/v1/`, `/v2/`)
- Manual DAG definition still supported
- Gradual migration of users to auto-discovery

### 17.2 Rollout Plan
1. **Alpha**: Internal testing with one production service
2. **Beta**: Gradual rollout to 10% of users, with manual DAG override option
3. **GA**: Full rollout, auto-discovery as default
4. **Deprecation**: Manual DAG definition deprecated in 12 months

## 18. Success Metrics

- Topology accuracy >99% vs. manual definitions
- 95% reduction in manual DAG updates
- <5 second detection latency for new services
- <2% false positive change detection rate
- 99.9% availability of topology APIs
- Ingestion throughput >100k spans/sec

## 19. References

- [[OpenTelemetry Specification](https://opentelemetry.io/docs/specs/otel/)](https://opentelemetry.io/docs/specs/otel/)
- [[OpenTelemetry Semantic Conventions](https://opentelemetry.io/docs/specs/semconv/)](https://opentelemetry.io/docs/specs/semconv/)
- [[OTLP Protocol](https://github.com/open-telemetry/opentelemetry-proto)](https://github.com/open-telemetry/opentelemetry-proto)
- [[Graph Theory - Cycle Detection](https://en.wikipedia.org/wiki/Cycle_(graph_theory))](https://en.wikipedia.org/wiki/Cycle_(graph_theory))

Error	Handling
Malformed trace	Log warning, skip span, increment error counter
Missing service.name	Use span.peer_service or rpc.service, fallback to "unknown-service"
Inconsistent versions	Use most recent version seen
Dropped due to backpressure	Emit metric, return HTTP 429

Error	Handling
Detected cycle	Log warning, remove weakest edge, flag for review
Topology explosion (>5000 nodes)	Aggregate by domain, emit alert
Failed change detection	Skip version, retry on next interval

Automatically ingest trace topology (e.g., from OpenTelemetry) to build and update these DAGs dynamically #2

Description

Feature Specification: Automatic Trace Topology Ingestion & Dynamic DAG Building

1. Overview

2. Problem Statement

3. Goals

4. Scope

In Scope

Out of Scope

5. Technical Architecture

5.1 Core Components

5.2 Data Flow

6. Data Model

6.1 Key Span Attributes (OpenTelemetry Semantic Conventions)

6.2 Service Node

6.3 Edge (Call Relationship)

6.4 DAG Version

7. Functional Requirements

FR1: Trace Ingestion

FR2: Service Discovery

FR3: Call Relationship Detection

FR4: DAG Construction

FR5: Change Detection & Notification

FR6: Query & Export APIs

FR7: Aggregation Window

8. Non-Functional Requirements

NFR1: Performance

NFR2: Reliability

NFR3: Scalability

NFR4: Data Quality

NFR5: Storage

9. API Specification

9.1 Trace Ingestion Endpoint

9.2 Topology Query APIs

Get service details

Get call relationships

Get DAG for visualization

List DAG versions

Get specific DAG version

Get topology changes

9.3 Configuration API

Define service kind overrides

10. Implementation Details

10.1 Technology Stack

10.2 Topology Discovery Algorithm

10.3 Cycle Detection & Resolution

10.4 Change Detection Algorithm

11. Configuration

11.1 Environment Variables

Topology discovery

Change detection

Storage

Performance

11.2 Configuration File Example

12. Example Usage

12.1 Sending Traces

Configure OTLP exporter pointing to NeuralBudget

Spans automatically contribute to topology discovery

12.2 Querying Topology

Export as GraphML for visualization

Check what changed in last hour

Get service dependencies

13. Error Handling

13.1 Trace Processing Errors

14. Testing Strategy

14.1 Unit Tests

14.2 Integration Tests

14.3 Load Testing

14.4 Test Data

15. Monitoring & Observability

15.1 Key Metrics

Topology discovery

Storage

15.2 Alerting Rules

16. Future Enhancements

17. Migration & Rollout

17.1 Backward Compatibility

17.2 Rollout Plan

18. Success Metrics

19. References