mirror of
https://github.com/andwati/telecom-platform.git
synced 2026-11-04 10:52:19 +00:00
Revert "App layer"
This commit is contained in:
Nutcase
@@ -1,333 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
var (
|
||||
// HTTP request metrics
|
||||
httpRequestsTotal = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_http_requests_total",
|
||||
Help: "Total number of HTTP requests",
|
||||
},
|
||||
[]string{"method", "endpoint", "status_code"},
|
||||
)
|
||||
|
||||
httpRequestDuration = prometheus.NewHistogramVec(
|
||||
prometheus.HistogramOpts{
|
||||
Name: "telecom_http_request_duration_seconds",
|
||||
Help: "HTTP request duration in seconds",
|
||||
Buckets: prometheus.DefBuckets,
|
||||
},
|
||||
[]string{"method", "endpoint"},
|
||||
)
|
||||
|
||||
// Subscriber metrics
|
||||
subscribersTotal = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_subscribers_total",
|
||||
Help: "Total number of subscribers",
|
||||
},
|
||||
)
|
||||
|
||||
subscribersActive = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_subscribers_active",
|
||||
Help: "Number of active subscribers",
|
||||
},
|
||||
)
|
||||
|
||||
subscribersSuspended = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_subscribers_suspended",
|
||||
Help: "Number of suspended subscribers",
|
||||
},
|
||||
)
|
||||
|
||||
// Usage metrics
|
||||
dataUsageBytes = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_data_usage_bytes_total",
|
||||
Help: "Total data usage in bytes",
|
||||
},
|
||||
[]string{"subscriber_id", "direction"},
|
||||
)
|
||||
|
||||
voiceUsageSeconds = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_voice_usage_seconds_total",
|
||||
Help: "Total voice usage in seconds",
|
||||
},
|
||||
[]string{"subscriber_id"},
|
||||
)
|
||||
|
||||
smsUsageTotal = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_sms_usage_total",
|
||||
Help: "Total SMS count",
|
||||
},
|
||||
[]string{"subscriber_id"},
|
||||
)
|
||||
|
||||
// Charging metrics
|
||||
creditBalance = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_credit_balance",
|
||||
Help: "Subscriber credit balance",
|
||||
},
|
||||
[]string{"subscriber_id"},
|
||||
)
|
||||
|
||||
paymentTransactionsTotal = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_payment_transactions_total",
|
||||
Help: "Total payment transactions",
|
||||
},
|
||||
[]string{"status", "gateway"},
|
||||
)
|
||||
|
||||
// System metrics
|
||||
activeSessions = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_active_sessions",
|
||||
Help: "Number of active sessions",
|
||||
},
|
||||
)
|
||||
|
||||
systemUptime = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_system_uptime_seconds",
|
||||
Help: "System uptime in seconds",
|
||||
},
|
||||
)
|
||||
|
||||
// Database metrics
|
||||
dbConnectionsActive = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_db_connections_active",
|
||||
Help: "Number of active database connections",
|
||||
},
|
||||
)
|
||||
|
||||
dbQueryDuration = prometheus.NewHistogramVec(
|
||||
prometheus.HistogramOpts{
|
||||
Name: "telecom_db_query_duration_seconds",
|
||||
Help: "Database query duration in seconds",
|
||||
Buckets: prometheus.DefBuckets,
|
||||
},
|
||||
[]string{"operation"},
|
||||
)
|
||||
|
||||
// Redis metrics
|
||||
redisConnectionsActive = prometheus.NewGauge(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "telecom_redis_connections_active",
|
||||
Help: "Number of active Redis connections",
|
||||
},
|
||||
)
|
||||
|
||||
redisOperationsTotal = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_redis_operations_total",
|
||||
Help: "Total Redis operations",
|
||||
},
|
||||
[]string{"operation", "status"},
|
||||
)
|
||||
|
||||
// Chaos engineering metrics
|
||||
chaosExperimentsTotal = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_chaos_experiments_total",
|
||||
Help: "Total chaos experiments run",
|
||||
},
|
||||
[]string{"type", "status"},
|
||||
)
|
||||
|
||||
// eBPF packet gateway metrics
|
||||
packetsProcessed = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_packets_processed_total",
|
||||
Help: "Total packets processed by eBPF gateway",
|
||||
},
|
||||
[]string{"action"},
|
||||
)
|
||||
|
||||
packetsDropped = prometheus.NewCounterVec(
|
||||
prometheus.CounterOpts{
|
||||
Name: "telecom_packets_dropped_total",
|
||||
Help: "Total packets dropped by eBPF gateway",
|
||||
},
|
||||
[]string{"reason"},
|
||||
)
|
||||
)
|
||||
|
||||
// MetricsCollector handles Prometheus metrics collection
|
||||
type MetricsCollector struct {
|
||||
startTime time.Time
|
||||
}
|
||||
|
||||
// NewMetricsCollector creates a new metrics collector
|
||||
func NewMetricsCollector() *MetricsCollector {
|
||||
return &MetricsCollector{
|
||||
startTime: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
// Register registers all metrics with Prometheus
|
||||
func (mc *MetricsCollector) Register() {
|
||||
prometheus.MustRegister(httpRequestsTotal)
|
||||
prometheus.MustRegister(httpRequestDuration)
|
||||
prometheus.MustRegister(subscribersTotal)
|
||||
prometheus.MustRegister(subscribersActive)
|
||||
prometheus.MustRegister(subscribersSuspended)
|
||||
prometheus.MustRegister(dataUsageBytes)
|
||||
prometheus.MustRegister(voiceUsageSeconds)
|
||||
prometheus.MustRegister(smsUsageTotal)
|
||||
prometheus.MustRegister(creditBalance)
|
||||
prometheus.MustRegister(paymentTransactionsTotal)
|
||||
prometheus.MustRegister(activeSessions)
|
||||
prometheus.MustRegister(systemUptime)
|
||||
prometheus.MustRegister(dbConnectionsActive)
|
||||
prometheus.MustRegister(dbQueryDuration)
|
||||
prometheus.MustRegister(redisConnectionsActive)
|
||||
prometheus.MustRegister(redisOperationsTotal)
|
||||
prometheus.MustRegister(chaosExperimentsTotal)
|
||||
prometheus.MustRegister(packetsProcessed)
|
||||
prometheus.MustRegister(packetsDropped)
|
||||
|
||||
// Start system uptime goroutine
|
||||
go mc.updateSystemMetrics()
|
||||
}
|
||||
|
||||
// updateSystemMetrics updates system-level metrics periodically
|
||||
func (mc *MetricsCollector) updateSystemMetrics() {
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
|
||||
for range ticker.C {
|
||||
systemUptime.Set(time.Since(mc.startTime).Seconds())
|
||||
}
|
||||
}
|
||||
|
||||
// HTTPMiddleware returns middleware to track HTTP metrics
|
||||
func (mc *MetricsCollector) HTTPMiddleware(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
start := time.Now()
|
||||
|
||||
// Create a response writer to capture status code
|
||||
rw := &responseWriter{ResponseWriter: w, statusCode: 200}
|
||||
|
||||
// Process request
|
||||
next.ServeHTTP(rw, r)
|
||||
|
||||
// Record metrics
|
||||
duration := time.Since(start).Seconds()
|
||||
statusCode := rw.statusCode
|
||||
|
||||
httpRequestsTotal.WithLabelValues(r.Method, r.URL.Path, http.StatusText(statusCode)).Inc()
|
||||
httpRequestDuration.WithLabelValues(r.Method, r.URL.Path).Observe(duration)
|
||||
})
|
||||
}
|
||||
|
||||
// responseWriter is a wrapper to capture status code
|
||||
type responseWriter struct {
|
||||
http.ResponseWriter
|
||||
statusCode int
|
||||
}
|
||||
|
||||
func (rw *responseWriter) WriteHeader(code int) {
|
||||
rw.statusCode = code
|
||||
rw.ResponseWriter.WriteHeader(code)
|
||||
}
|
||||
|
||||
// UpdateSubscriberMetrics updates subscriber-related metrics
|
||||
func (mc *MetricsCollector) UpdateSubscriberMetrics(total, active, suspended int) {
|
||||
subscribersTotal.Set(float64(total))
|
||||
subscribersActive.Set(float64(active))
|
||||
subscribersSuspended.Set(float64(suspended))
|
||||
}
|
||||
|
||||
// RecordDataUsage records data usage for a subscriber
|
||||
func (mc *MetricsCollector) RecordDataUsage(subscriberID string, direction string, bytes int64) {
|
||||
dataUsageBytes.WithLabelValues(subscriberID, direction).Add(float64(bytes))
|
||||
}
|
||||
|
||||
// RecordVoiceUsage records voice usage for a subscriber
|
||||
func (mc *MetricsCollector) RecordVoiceUsage(subscriberID string, seconds int64) {
|
||||
voiceUsageSeconds.WithLabelValues(subscriberID).Add(float64(seconds))
|
||||
}
|
||||
|
||||
// RecordSMSUsage records SMS usage for a subscriber
|
||||
func (mc *MetricsCollector) RecordSMSUsage(subscriberID string, count int64) {
|
||||
smsUsageTotal.WithLabelValues(subscriberID).Add(float64(count))
|
||||
}
|
||||
|
||||
// UpdateCreditBalance updates credit balance for a subscriber
|
||||
func (mc *MetricsCollector) UpdateCreditBalance(subscriberID string, balance float64) {
|
||||
creditBalance.WithLabelValues(subscriberID).Set(balance)
|
||||
}
|
||||
|
||||
// RecordPaymentTransaction records a payment transaction
|
||||
func (mc *MetricsCollector) RecordPaymentTransaction(status, gateway string) {
|
||||
paymentTransactionsTotal.WithLabelValues(status, gateway).Inc()
|
||||
}
|
||||
|
||||
// UpdateActiveSessions updates the active sessions count
|
||||
func (mc *MetricsCollector) UpdateActiveSessions(count int) {
|
||||
activeSessions.Set(float64(count))
|
||||
}
|
||||
|
||||
// UpdateDBMetrics updates database metrics
|
||||
func (mc *MetricsCollector) UpdateDBMetrics(activeConnections int, operation string, duration time.Duration) {
|
||||
dbConnectionsActive.Set(float64(activeConnections))
|
||||
dbQueryDuration.WithLabelValues(operation).Observe(duration.Seconds())
|
||||
}
|
||||
|
||||
// UpdateRedisMetrics updates Redis metrics
|
||||
func (mc *MetricsCollector) UpdateRedisMetrics(activeConnections int, operation, status string) {
|
||||
redisConnectionsActive.Set(float64(activeConnections))
|
||||
redisOperationsTotal.WithLabelValues(operation, status).Inc()
|
||||
}
|
||||
|
||||
// RecordChaosExperiment records a chaos experiment
|
||||
func (mc *MetricsCollector) RecordChaosExperiment(experimentType, status string) {
|
||||
chaosExperimentsTotal.WithLabelValues(experimentType, status).Inc()
|
||||
}
|
||||
|
||||
// RecordPacketMetrics records packet processing metrics
|
||||
func (mc *MetricsCollector) RecordPacketMetrics(action string, count int64, reason string) {
|
||||
if action == "dropped" {
|
||||
packetsDropped.WithLabelValues(reason).Add(float64(count))
|
||||
} else {
|
||||
packetsProcessed.WithLabelValues(action).Add(float64(count))
|
||||
}
|
||||
}
|
||||
|
||||
// Handler returns the Prometheus metrics HTTP handler
|
||||
func (mc *MetricsCollector) Handler() http.Handler {
|
||||
return promhttp.Handler()
|
||||
}
|
||||
|
||||
// StartMetricsServer starts the Prometheus metrics server
|
||||
func (mc *MetricsCollector) StartMetricsServer(ctx context.Context, addr string) error {
|
||||
mux := http.NewServeMux()
|
||||
mux.Handle("/metrics", mc.Handler())
|
||||
|
||||
server := &http.Server{
|
||||
Addr: addr,
|
||||
Handler: mux,
|
||||
}
|
||||
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
server.Shutdown(context.Background())
|
||||
}()
|
||||
|
||||
return server.ListenAndServe()
|
||||
}
|
||||
@@ -1,387 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
influxdb2 "github.com/influxdata/influxdb-client-go/v2"
|
||||
"github.com/influxdata/influxdb-client-go/v2/api"
|
||||
)
|
||||
|
||||
// TimeSeriesStorage handles advanced time-series metrics storage
|
||||
type TimeSeriesStorage struct {
|
||||
client api.WriteAPIBlocking
|
||||
queryAPI api.QueryAPI
|
||||
org string
|
||||
bucket string
|
||||
}
|
||||
|
||||
// MetricPoint represents a single metric data point
|
||||
type MetricPoint struct {
|
||||
Measurement string `json:"measurement"`
|
||||
Tags map[string]string `json:"tags"`
|
||||
Fields map[string]any `json:"fields"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// UsageMetrics represents usage metrics for time-series storage
|
||||
type UsageMetrics struct {
|
||||
SubscriberID string `json:"subscriber_id"`
|
||||
IMSI string `json:"imsi"`
|
||||
DataUp int64 `json:"data_up"`
|
||||
DataDown int64 `json:"data_down"`
|
||||
VoiceSeconds int64 `json:"voice_seconds"`
|
||||
SMSCount int64 `json:"sms_count"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// SystemMetrics represents system metrics for time-series storage
|
||||
type SystemMetrics struct {
|
||||
ActiveSessions int `json:"active_sessions"`
|
||||
CPUUsage float64 `json:"cpu_usage"`
|
||||
MemoryUsage float64 `json:"memory_usage"`
|
||||
NetworkRX int64 `json:"network_rx"`
|
||||
NetworkTX int64 `json:"network_tx"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// ChargingMetrics represents charging metrics for time-series storage
|
||||
type ChargingMetrics struct {
|
||||
SubscriberID string `json:"subscriber_id"`
|
||||
Balance float64 `json:"balance"`
|
||||
TransactionID string `json:"transaction_id"`
|
||||
Amount float64 `json:"amount"`
|
||||
Currency string `json:"currency"`
|
||||
Status string `json:"status"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
}
|
||||
|
||||
// NewTimeSeriesStorage creates a new time-series storage client
|
||||
func NewTimeSeriesStorage(url, token, org, bucket string) (*TimeSeriesStorage, error) {
|
||||
client := influxdb2.NewClient(url, token)
|
||||
writeAPI := client.WriteAPIBlocking(org, bucket)
|
||||
queryAPI := client.QueryAPI(org)
|
||||
|
||||
return &TimeSeriesStorage{
|
||||
client: writeAPI,
|
||||
queryAPI: queryAPI,
|
||||
org: org,
|
||||
bucket: bucket,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// StoreUsageMetrics stores usage metrics in InfluxDB
|
||||
func (ts *TimeSeriesStorage) StoreUsageMetrics(ctx context.Context, metrics UsageMetrics) error {
|
||||
point := influxdb2.NewPoint(
|
||||
"usage",
|
||||
map[string]string{
|
||||
"subscriber_id": metrics.SubscriberID,
|
||||
"imsi": metrics.IMSI,
|
||||
},
|
||||
map[string]interface{}{
|
||||
"data_up": metrics.DataUp,
|
||||
"data_down": metrics.DataDown,
|
||||
"voice_seconds": metrics.VoiceSeconds,
|
||||
"sms_count": metrics.SMSCount,
|
||||
},
|
||||
metrics.Timestamp,
|
||||
)
|
||||
|
||||
return ts.client.WritePoint(ctx, point)
|
||||
}
|
||||
|
||||
// StoreSystemMetrics stores system metrics in InfluxDB
|
||||
func (ts *TimeSeriesStorage) StoreSystemMetrics(ctx context.Context, metrics SystemMetrics) error {
|
||||
point := influxdb2.NewPoint(
|
||||
"system",
|
||||
map[string]string{},
|
||||
map[string]interface{}{
|
||||
"active_sessions": metrics.ActiveSessions,
|
||||
"cpu_usage": metrics.CPUUsage,
|
||||
"memory_usage": metrics.MemoryUsage,
|
||||
"network_rx": metrics.NetworkRX,
|
||||
"network_tx": metrics.NetworkTX,
|
||||
},
|
||||
metrics.Timestamp,
|
||||
)
|
||||
|
||||
return ts.client.WritePoint(ctx, point)
|
||||
}
|
||||
|
||||
// StoreChargingMetrics stores charging metrics in InfluxDB
|
||||
func (ts *TimeSeriesStorage) StoreChargingMetrics(ctx context.Context, metrics ChargingMetrics) error {
|
||||
point := influxdb2.NewPoint(
|
||||
"charging",
|
||||
map[string]string{
|
||||
"subscriber_id": metrics.SubscriberID,
|
||||
"transaction_id": metrics.TransactionID,
|
||||
"currency": metrics.Currency,
|
||||
"status": metrics.Status,
|
||||
},
|
||||
map[string]interface{}{
|
||||
"balance": metrics.Balance,
|
||||
"amount": metrics.Amount,
|
||||
},
|
||||
metrics.Timestamp,
|
||||
)
|
||||
|
||||
return ts.client.WritePoint(ctx, point)
|
||||
}
|
||||
|
||||
// StorePacketMetrics stores packet processing metrics in InfluxDB
|
||||
func (ts *TimeSeriesStorage) StorePacketMetrics(ctx context.Context, subscriberID string, packetsProcessed, packetsDropped int64, bytesProcessed int64) error {
|
||||
point := influxdb2.NewPoint(
|
||||
"packets",
|
||||
map[string]string{
|
||||
"subscriber_id": subscriberID,
|
||||
},
|
||||
map[string]interface{}{
|
||||
"packets_processed": packetsProcessed,
|
||||
"packets_dropped": packetsDropped,
|
||||
"bytes_processed": bytesProcessed,
|
||||
},
|
||||
time.Now(),
|
||||
)
|
||||
|
||||
return ts.client.WritePoint(ctx, point)
|
||||
}
|
||||
|
||||
// GetUsageStats retrieves usage statistics for a time range
|
||||
func (ts *TimeSeriesStorage) GetUsageStats(ctx context.Context, subscriberID string, start, end time.Time) (*UsageMetrics, error) {
|
||||
query := fmt.Sprintf(`
|
||||
from(bucket: "%s")
|
||||
|> range(start: %s, stop: %s)
|
||||
|> filter(fn: (r) => r._measurement == "usage")
|
||||
|> filter(fn: (r) => r.subscriber_id == "%s")
|
||||
|> last()
|
||||
`, ts.bucket, start.Format(time.RFC3339), end.Format(time.RFC3339), subscriberID)
|
||||
|
||||
result, err := ts.queryAPI.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query usage stats: %w", err)
|
||||
}
|
||||
|
||||
var metrics UsageMetrics
|
||||
for result.Next() {
|
||||
record := result.Record()
|
||||
metrics = UsageMetrics{
|
||||
SubscriberID: record.ValueByKey("subscriber_id").(string),
|
||||
IMSI: record.ValueByKey("imsi").(string),
|
||||
DataUp: record.ValueByKey("data_up").(int64),
|
||||
DataDown: record.ValueByKey("data_down").(int64),
|
||||
VoiceSeconds: record.ValueByKey("voice_seconds").(int64),
|
||||
SMSCount: record.ValueByKey("sms_count").(int64),
|
||||
Timestamp: record.Time(),
|
||||
}
|
||||
}
|
||||
|
||||
if result.Err() != nil {
|
||||
return nil, result.Err()
|
||||
}
|
||||
|
||||
return &metrics, nil
|
||||
}
|
||||
|
||||
// GetSystemMetricsHistory retrieves system metrics history
|
||||
func (ts *TimeSeriesStorage) GetSystemMetricsHistory(ctx context.Context, start, end time.Time, interval string) ([]SystemMetrics, error) {
|
||||
query := fmt.Sprintf(`
|
||||
from(bucket: "%s")
|
||||
|> range(start: %s, stop: %s)
|
||||
|> filter(fn: (r) => r._measurement == "system")
|
||||
|> aggregateWindow(every: %s, fn: mean, createEmpty: false)
|
||||
|> yield(name: "mean")
|
||||
`, ts.bucket, start.Format(time.RFC3339), end.Format(time.RFC3339), interval)
|
||||
|
||||
result, err := ts.queryAPI.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query system metrics: %w", err)
|
||||
}
|
||||
|
||||
var metrics []SystemMetrics
|
||||
for result.Next() {
|
||||
record := result.Record()
|
||||
metric := SystemMetrics{
|
||||
ActiveSessions: int(record.ValueByKey("active_sessions").(int64)),
|
||||
CPUUsage: record.ValueByKey("cpu_usage").(float64),
|
||||
MemoryUsage: record.ValueByKey("memory_usage").(float64),
|
||||
NetworkRX: record.ValueByKey("network_rx").(int64),
|
||||
NetworkTX: record.ValueByKey("network_tx").(int64),
|
||||
Timestamp: record.Time(),
|
||||
}
|
||||
metrics = append(metrics, metric)
|
||||
}
|
||||
|
||||
if result.Err() != nil {
|
||||
return nil, result.Err()
|
||||
}
|
||||
|
||||
return metrics, nil
|
||||
}
|
||||
|
||||
// GetSubscriberUsageHistory retrieves usage history for a subscriber
|
||||
func (ts *TimeSeriesStorage) GetSubscriberUsageHistory(ctx context.Context, subscriberID string, start, end time.Time, interval string) ([]UsageMetrics, error) {
|
||||
query := fmt.Sprintf(`
|
||||
from(bucket: "%s")
|
||||
|> range(start: %s, stop: %s)
|
||||
|> filter(fn: (r) => r._measurement == "usage")
|
||||
|> filter(fn: (r) => r.subscriber_id == "%s")
|
||||
|> aggregateWindow(every: %s, fn: sum, createEmpty: false)
|
||||
|> yield(name: "sum")
|
||||
`, ts.bucket, start.Format(time.RFC3339), end.Format(time.RFC3339), subscriberID, interval)
|
||||
|
||||
result, err := ts.queryAPI.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query subscriber usage: %w", err)
|
||||
}
|
||||
|
||||
var metrics []UsageMetrics
|
||||
for result.Next() {
|
||||
record := result.Record()
|
||||
metric := UsageMetrics{
|
||||
SubscriberID: subscriberID,
|
||||
IMSI: record.ValueByKey("imsi").(string),
|
||||
DataUp: record.ValueByKey("data_up").(int64),
|
||||
DataDown: record.ValueByKey("data_down").(int64),
|
||||
VoiceSeconds: record.ValueByKey("voice_seconds").(int64),
|
||||
SMSCount: record.ValueByKey("sms_count").(int64),
|
||||
Timestamp: record.Time(),
|
||||
}
|
||||
metrics = append(metrics, metric)
|
||||
}
|
||||
|
||||
if result.Err() != nil {
|
||||
return nil, result.Err()
|
||||
}
|
||||
|
||||
return metrics, nil
|
||||
}
|
||||
|
||||
// GetChargingHistory retrieves charging history for a subscriber
|
||||
func (ts *TimeSeriesStorage) GetChargingHistory(ctx context.Context, subscriberID string, start, end time.Time) ([]ChargingMetrics, error) {
|
||||
query := fmt.Sprintf(`
|
||||
from(bucket: "%s")
|
||||
|> range(start: %s, stop: %s)
|
||||
|> filter(fn: (r) => r._measurement == "charging")
|
||||
|> filter(fn: (r) => r.subscriber_id == "%s")
|
||||
|> sort(columns: ["_time"])
|
||||
`, ts.bucket, start.Format(time.RFC3339), end.Format(time.RFC3339), subscriberID)
|
||||
|
||||
result, err := ts.queryAPI.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query charging history: %w", err)
|
||||
}
|
||||
|
||||
var metrics []ChargingMetrics
|
||||
for result.Next() {
|
||||
record := result.Record()
|
||||
metric := ChargingMetrics{
|
||||
SubscriberID: record.ValueByKey("subscriber_id").(string),
|
||||
TransactionID: record.ValueByKey("transaction_id").(string),
|
||||
Balance: record.ValueByKey("balance").(float64),
|
||||
Amount: record.ValueByKey("amount").(float64),
|
||||
Currency: record.ValueByKey("currency").(string),
|
||||
Status: record.ValueByKey("status").(string),
|
||||
Timestamp: record.Time(),
|
||||
}
|
||||
metrics = append(metrics, metric)
|
||||
}
|
||||
|
||||
if result.Err() != nil {
|
||||
return nil, result.Err()
|
||||
}
|
||||
|
||||
return metrics, nil
|
||||
}
|
||||
|
||||
// GetTopDataUsers retrieves top data users for a time period
|
||||
func (ts *TimeSeriesStorage) GetTopDataUsers(ctx context.Context, start, end time.Time, limit int) ([]map[string]interface{}, error) {
|
||||
query := fmt.Sprintf(`
|
||||
from(bucket: "%s")
|
||||
|> range(start: %s, stop: %s)
|
||||
|> filter(fn: (r) => r._measurement == "usage")
|
||||
|> group(columns: ["subscriber_id"])
|
||||
|> sum(columns: ["data_up", "data_down"])
|
||||
|> map(fn: (r) => ({r with total_data: r.data_up + r.data_down}))
|
||||
|> sort(columns: ["total_data"], desc: true)
|
||||
|> limit(n: %d)
|
||||
`, ts.bucket, start.Format(time.RFC3339), end.Format(time.RFC3339), limit)
|
||||
|
||||
result, err := ts.queryAPI.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query top data users: %w", err)
|
||||
}
|
||||
|
||||
var users []map[string]interface{}
|
||||
for result.Next() {
|
||||
record := result.Record()
|
||||
user := map[string]interface{}{
|
||||
"subscriber_id": record.ValueByKey("subscriber_id"),
|
||||
"total_data": record.ValueByKey("total_data"),
|
||||
"data_up": record.ValueByKey("data_up"),
|
||||
"data_down": record.ValueByKey("data_down"),
|
||||
}
|
||||
users = append(users, user)
|
||||
}
|
||||
|
||||
if result.Err() != nil {
|
||||
return nil, result.Err()
|
||||
}
|
||||
|
||||
return users, nil
|
||||
}
|
||||
|
||||
// Close closes the InfluxDB client connection
|
||||
func (ts *TimeSeriesStorage) Close() {
|
||||
// InfluxDB client doesn't need explicit closing in v2
|
||||
}
|
||||
|
||||
// MetricsCollectorWithTimeSeries extends MetricsCollector with time-series storage
|
||||
type MetricsCollectorWithTimeSeries struct {
|
||||
*MetricsCollector
|
||||
timeSeries *TimeSeriesStorage
|
||||
}
|
||||
|
||||
// NewMetricsCollectorWithTimeSeries creates a new metrics collector with time-series storage
|
||||
func NewMetricsCollectorWithTimeSeries(prometheusCollector *MetricsCollector, timeSeries *TimeSeriesStorage) *MetricsCollectorWithTimeSeries {
|
||||
return &MetricsCollectorWithTimeSeries{
|
||||
MetricsCollector: prometheusCollector,
|
||||
timeSeries: timeSeries,
|
||||
}
|
||||
}
|
||||
|
||||
// StoreUsageMetricsInTimeSeries stores usage metrics in both Prometheus and time-series
|
||||
func (mc *MetricsCollectorWithTimeSeries) StoreUsageMetricsInTimeSeries(ctx context.Context, metrics UsageMetrics) error {
|
||||
// Update Prometheus metrics
|
||||
mc.RecordDataUsage(metrics.SubscriberID, "up", metrics.DataUp)
|
||||
mc.RecordDataUsage(metrics.SubscriberID, "down", metrics.DataDown)
|
||||
mc.RecordVoiceUsage(metrics.SubscriberID, metrics.VoiceSeconds)
|
||||
mc.RecordSMSUsage(metrics.SubscriberID, metrics.SMSCount)
|
||||
|
||||
// Store in time-series
|
||||
return mc.timeSeries.StoreUsageMetrics(ctx, metrics)
|
||||
}
|
||||
|
||||
// StoreSystemMetricsInTimeSeries stores system metrics in both Prometheus and time-series
|
||||
func (mc *MetricsCollectorWithTimeSeries) StoreSystemMetricsInTimeSeries(ctx context.Context, metrics SystemMetrics) error {
|
||||
// Update Prometheus metrics
|
||||
mc.UpdateActiveSessions(metrics.ActiveSessions)
|
||||
|
||||
// Store in time-series
|
||||
return mc.timeSeries.StoreSystemMetrics(ctx, metrics)
|
||||
}
|
||||
|
||||
// StoreChargingMetricsInTimeSeries stores charging metrics in both Prometheus and time-series
|
||||
func (mc *MetricsCollectorWithTimeSeries) StoreChargingMetricsInTimeSeries(ctx context.Context, metrics ChargingMetrics) error {
|
||||
// Update Prometheus metrics
|
||||
mc.UpdateCreditBalance(metrics.SubscriberID, metrics.Balance)
|
||||
mc.RecordPaymentTransaction(metrics.Status, "stripe")
|
||||
|
||||
// Store in time-series
|
||||
return mc.timeSeries.StoreChargingMetrics(ctx, metrics)
|
||||
}
|
||||
|
||||
// GetTimeSeriesStorage returns the time-series storage instance
|
||||
func (mc *MetricsCollectorWithTimeSeries) GetTimeSeriesStorage() *TimeSeriesStorage {
|
||||
return mc.timeSeries
|
||||
}
|
||||
Reference in New Issue
Block a user