Merge pull request #589 from sahandilshan/analytics

Add initial observability component

Author: sahandilshan

backend/cmd/server/main.go

@@ -31,11 +31,13 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/asgardeo/thunder/internal/observability"
 	"github.com/asgardeo/thunder/internal/system/cache"
 	"github.com/asgardeo/thunder/internal/system/cert"
 	"github.com/asgardeo/thunder/internal/system/config"
 	"github.com/asgardeo/thunder/internal/system/database/provider"
 	"github.com/asgardeo/thunder/internal/system/log"
+	"github.com/asgardeo/thunder/internal/system/middleware"
 )
 
 // shutdownTimeout defines the timeout duration for graceful shutdown.
@@ -54,6 +56,9 @@ func main() {
 	// Initialize the cache manager.
 	initCacheManager(logger)
 
+	// Initialize observability with console adapter and JSON format
+	initObservability(logger)
+
 	// Create a new HTTP multiplexer.
 	mux := http.NewServeMux()
 	if mux == nil {
@@ -136,6 +141,34 @@ func initCacheManager(logger *log.Logger) {
 	cm.Init()
 }
 
+// initObservability initializes the observability service with console adapter and JSON format.
+func initObservability(logger *log.Logger) {
+	// Configure observability to use console adapter with JSON format
+	observabilityCfg := &observability.Config{
+		Enabled: true,
+		Output: observability.OutputConfig{
+			Type:   "console", // Output to stdout
+			Format: "json",
+		},
+		Metrics: observability.MetricsConfig{
+			Enabled: true,
+		},
+		FailureMode: "graceful", // Don't fail if observability has issues
+	}
+
+	svc, err := observability.InitializeWithConfig(observabilityCfg)
+	if err != nil {
+		logger.Error("Failed to initialize observability service", log.Error(err))
+		return
+	}
+
+	if svc.IsEnabled() {
+		logger.Debug("Observability service initialized successfully with console adapter and JSON format")
+	} else {
+		logger.Warn("Observability service is disabled")
+	}
+}
+
 // loadCertConfig loads the certificate configuration and extracts the Key ID (kid).
 func loadCertConfig(logger *log.Logger, cfg *config.Config, thunderHome string) *tls.Config {
 	sysCertSvc := cert.NewSystemCertificateService()
@@ -198,15 +231,15 @@ func startHTTPServer(logger *log.Logger, cfg *config.Config, mux *http.ServeMux)
 
 // createHTTPServer creates and configures an HTTP server with common settings.
 func createHTTPServer(logger *log.Logger, cfg *config.Config, mux *http.ServeMux) (*http.Server, string) {
-	// Wrap the multiplexer with AccessLogHandler.
-	wrappedMux := log.AccessLogHandler(logger, mux)
+	handler := middleware.CorrelationIDMiddleware(mux)
+	handler = log.AccessLogHandler(logger, handler)
 
 	// Build the server address using hostname and port from the configurations.
 	serverAddr := fmt.Sprintf("%s:%d", cfg.Server.Hostname, cfg.Server.Port)
 
 	server := &http.Server{
 		Addr:              serverAddr,
-		Handler:           wrappedMux,
+		Handler:           handler,
 		ReadHeaderTimeout: 10 * time.Second, // Mitigate Slowloris attacks
 		WriteTimeout:      10 * time.Second,
 		IdleTimeout:       120 * time.Second,
@@ -227,6 +260,13 @@ func gracefulShutdown(logger *log.Logger, server *http.Server) {
 		logger.Debug("HTTP server shutdown completed")
 	}
 
+	// Shutdown observability service
+	observabilitySvc := observability.GetService()
+	if observabilitySvc != nil {
+		observabilitySvc.Shutdown()
+		logger.Debug("Observability service shutdown completed")
+	}
+
 	// Close database connections
 	dbCloser := provider.GetDBProviderCloser()
 	if err := dbCloser.Close(); err != nil {

backend/cmd/server/repository/conf/deployment.yaml

@@ -29,3 +29,12 @@ cors:
 
 immutable_resources:
   enabled: false
+
+observability:
+  enabled: false
+  output:
+    type: "console"
+    format: "json"
+  metrics:
+    enabled: true
+  failure_mode: "graceful"

backend/internal/observability/README.md

@@ -0,0 +1,258 @@
+# Observability System
+
+A comprehensive observability platform for Thunder providing event logging, metrics collection, distributed tracing, and health monitoring for authentication and authorization flows.
+
+## Architecture
+
+The observability system uses an **Event Bus** pattern to capture and process events throughout the auth/authz flows with minimal performance impact.
+
+**Key Features:**
+- ✅ **Async & Non-blocking** - Events published in goroutines, main thread never blocks
+- ✅ **No Queue** - Direct event delivery to subscribers
+- ✅ **Category-based routing** - Subscribers filter by event categories
+- ✅ **Isolated failures** - Subscriber panics don't affect others
+
+```
+Auth/AuthZ Components → Event Bus (Publisher) → Subscribers (async) → Formatters → Output Adapters
+                                    ↓
+                            Each subscriber in own goroutine
+```
+
+## Quick Start
+
+### 1. Publishing Events
+
+```go
+import (
+    "github.com/asgardeo/thunder/internal/observability"
+    "github.com/asgardeo/thunder/internal/observability/event"
+)
+
+func handleRequest(ctx context.Context) {
+    // Ensure trace ID exists
+    ctx = observability.EnsureTraceID(ctx)
+    traceID := observability.GetTraceID(ctx)
+
+    // Create and publish event
+    evt := event.NewEvent(traceID, string(event.EventTypeAuthenticationStarted), "MyComponent")
+    evt.WithData("user_id", "user123")
+       .WithStatus(event.StatusInProgress)
+       .WithData("message", "Authentication started")
+
+    observability.GetService().PublishEvent(evt)
+}
+```
+
+### 2. Custom Subscribers
+
+```go
+import (
+    "github.com/asgardeo/thunder/internal/observability/publisher"
+    "github.com/asgardeo/thunder/internal/observability/subscriber/defaultsubscriber"
+    "github.com/asgardeo/thunder/internal/observability/formatter/json"
+    "github.com/asgardeo/thunder/internal/observability/adapter/file"
+    "github.com/asgardeo/thunder/internal/observability/event"
+)
+
+// Subscribe to specific event types only
+jsonFormatter := jsonformatter.NewJSONFormatter()
+fileAdapter, _ := file.NewFileAdapter("/path/to/custom-events.log")
+subscriber := defaultsubscriber.NewDefaultSubscriber(jsonFormatter, fileAdapter)
+
+// Option 1: Subscribe to specific event types
+publisher.GetPublisher().Subscribe(subscriber,
+    event.EventTypeAuthenticationStarted,
+    event.EventTypeAuthenticationCompleted,
+    event.EventTypeAuthenticationFailed,
+)
+
+// Option 2: Subscribe to all events
+publisher.GetPublisher().SubscribeAll(subscriber)
+```
+
+## Package Structure
+
+```
+observability/
+├── event/                      # Event model and types
+│   ├── event.go               # Core event structure with fluent API
+│   └── constants.go           # Event types and component names
+├── publisher/                  # Event publisher
+│   └── publisher.go           # Singleton publisher with in-memory queue
+├── subscriber/                 # Subscriber interfaces and implementations
+│   ├── subscriber.go          # Subscriber interface
+│   └── defaultsubscriber/     # Default subscriber implementation
+│       └── default_subscriber.go
+├── formatter/                  # Event formatters
+│   ├── formatter.go           # Formatter interface
+│   ├── json/                  # JSON formatter
+│   │   └── json_formatter.go
+│   └── csv/                   # CSV formatter
+│       └── csv_formatter.go
+├── adapter/                    # Output adapters
+│   ├── adapter.go             # OutputAdapter interface
+│   ├── file/                  # File output
+│   │   └── file_adapter.go
+│   └── console/               # Console output
+│       └── console_adapter.go
+├── examples/                   # Integration examples
+│   └── integration_examples.go
+├── context.go                 # Context utilities for trace ID
+├── service.go                 # Main analytics service
+└── README.md                  # This file
+```
+
+## Event Types
+
+### Authorization Events
+- `AUTHORIZATION_STARTED` - Authorization request received
+- `AUTHORIZATION_VALIDATED` - Request validated
+- `AUTHORIZATION_CODE_GENERATED` - Auth code created
+- `AUTHORIZATION_COMPLETED` - Flow complete
+- `AUTHORIZATION_FAILED` - Flow failed
+
+### Authentication Events
+- `AUTHENTICATION_STARTED` - Auth flow begins
+- `CREDENTIALS_AUTH_STARTED` - Username/password auth
+- `CREDENTIALS_AUTH_COMPLETED` - Credentials verified
+- `CREDENTIALS_AUTH_FAILED` - Credentials invalid
+- `OTP_SENT` - OTP sent to user
+- `OTP_VERIFIED` - OTP validated
+- `SOCIAL_AUTH_STARTED` - Social login begins
+- `SOCIAL_AUTH_COMPLETED` - Social login succeeds
+- `AUTHENTICATION_COMPLETED` - Auth complete
+- `AUTHENTICATION_FAILED` - Auth failed
+
+### Token Events
+- `TOKEN_REQUEST_RECEIVED` - Token request received
+- `TOKEN_REQUEST_VALIDATED` - Request validated
+- `PKCE_VALIDATED` - PKCE validation successful
+- `ACCESS_TOKEN_GENERATED` - Access token created
+- `ID_TOKEN_GENERATED` - ID token created
+- `REFRESH_TOKEN_GENERATED` - Refresh token created
+- `TOKEN_ISSUED` - Tokens issued
+- `TOKEN_REQUEST_FAILED` - Request failed
+
+### Flow Events
+- `FLOW_STARTED` - Flow execution begins
+- `FLOW_NODE_EXECUTION_STARTED` - Node execution starts
+- `FLOW_NODE_EXECUTION_COMPLETED` - Node completes
+- `FLOW_USER_INPUT_REQUIRED` - User input needed
+- `FLOW_COMPLETED` - Flow succeeds
+- `FLOW_FAILED` - Flow fails
+
+## Integration Examples
+
+See `examples/integration_examples.go` for complete examples of:
+- Authorization handler integration
+- Token handler integration
+- Flow execution integration
+- Authentication service integration
+
+## Output Format
+
+### JSON (Default)
+```json
+{
+  "trace_id": "550e8400-e29b-41d4-a716-446655440000",
+  "event_id": "123e4567-e89b-12d3-a456-426614174000",
+  "event_type": "AUTHENTICATION_COMPLETED",
+  "timestamp": "2025-10-21T10:15:30Z",
+  "component": "AuthenticationService",
+  "user_id": "user_123",
+  "client_id": "client_456",
+  "status": "SUCCESS",
+  "duration_ms": 245
+}
+```
+
+### CSV
+```csv
+TraceID,EventID,EventType,Timestamp,Component,UserID,ClientID,Status,DurationMS
+550e8400...,123e4567...,AUTHENTICATION_COMPLETED,2025-10-21T10:15:30Z,AuthenticationService,user_123,client_456,SUCCESS,245
+```
+
+## Querying Events
+
+### Filter by Trace ID
+```bash
+# Using jq
+cat analytics.log | jq 'select(.trace_id == "550e8400-e29b-41d4-a716-446655440000")'
+
+# Get event timeline
+cat analytics.log | jq 'select(.trace_id == "550e8400...") | {timestamp, event_type, status}'
+```
+
+### Get Failed Authentications
+```bash
+cat analytics.log | jq 'select(.event_type | contains("FAILED"))'
+```
+
+### Calculate Average Duration
+```bash
+cat analytics.log | jq -s '[.[] | select(.event_type == "TOKEN_ISSUED") | .duration_ms] | add / length'
+```
+
+## Performance
+
+- **Non-blocking**: Event publishing returns immediately, never blocks the main thread
+- **Async processing**: Each subscriber runs in its own goroutine
+- **No queue overhead**: Direct event delivery without buffering
+- **Graceful degradation**: Events skipped if no subscribers are interested
+- **Parallel processing**: All subscribers process events simultaneously
+- **Isolated failures**: One subscriber's failure doesn't affect others
+
+## Extending
+
+### Custom Formatter
+
+```go
+type MyFormatter struct{}
+
+func (mf *MyFormatter) Format(evt *event.Event) ([]byte, error) {
+    // Custom formatting logic
+    return []byte(fmt.Sprintf("%s: %s", evt.EventType, evt.Message)), nil
+}
+
+func (mf *MyFormatter) GetName() string {
+    return "MyFormatter"
+}
+```
+
+### Custom Output Adapter
+
+```go
+type MyAdapter struct{}
+
+func (ma *MyAdapter) Write(data []byte) error {
+    // Custom output logic (send to external service, database, etc.)
+    return nil
+}
+
+func (ma *MyAdapter) Flush() error {
+    return nil
+}
+
+func (ma *MyAdapter) Close() error {
+    return nil
+}
+
+func (ma *MyAdapter) GetName() string {
+    return "MyAdapter"
+}
+```
+
+## Shutdown
+
+Always shut down the analytics service gracefully:
+
+```go
+// In main.go or shutdown handler
+defer observability.GetService().Shutdown()
+```
+
+This ensures:
+- Pending events are processed
+- Buffers are flushed
+- Files are closed properly
+- Resources are released

backend/internal/observability/adapter/adapter.go

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2025, WSO2 LLC. (https://www.wso2.com).
+ *
+ * WSO2 LLC. licenses this file to you under the Apache License,
+ * Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Package adapter provides output adapter interfaces and implementations.
+package adapter
+
+// OutputAdapter is the interface for writing formatted events to various destinations.
+type OutputAdapter interface {
+	// Write writes formatted event data to the output destination.
+	Write(data []byte) error
+
+	// Flush ensures all buffered data is written.
+	Flush() error
+
+	// Close closes the adapter and releases resources.
+	Close() error
+
+	// GetName returns the name of this adapter.
+	GetName() string
+}