cactoandClaude Sonnet 5
938700cf86
Subscribes to devices/+/telemetry, validates payloads, buffers writes into ClickHouse (flush on size or interval), and emits a per-reading event on RabbitMQ for rule-engine-service to consume later. Adds the initial Postgres and ClickHouse schema migrations, a Dockerfile, and wires the service into docker-compose. Verified end-to-end via docker compose + mosquitto_pub: row lands in ClickHouse and the event reaches the telemetry.new_reading queue. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
92 lines
2.3 KiB
Go
92 lines
2.3 KiB
Go
// Package config loads ingest-service settings from environment variables,
|
|
// matching the names used in the repo-root .env.example.
|
|
package config
|
|
|
|
import (
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"time"
|
|
)
|
|
|
|
type Config struct {
|
|
MQTTBrokerURL string
|
|
MQTTClientID string
|
|
MQTTTopic string
|
|
|
|
ClickHouseAddr string
|
|
ClickHouseDatabase string
|
|
ClickHouseUsername string
|
|
ClickHousePassword string
|
|
|
|
RabbitMQURL string
|
|
|
|
BatchMaxSize int
|
|
BatchFlushInterval time.Duration
|
|
BatchFlushTimeout time.Duration
|
|
}
|
|
|
|
func Load() (Config, error) {
|
|
cfg := Config{
|
|
MQTTBrokerURL: fmt.Sprintf("tcp://%s:%s", getEnv("MQTT_HOST", "localhost"), getEnv("MQTT_PORT", "1883")),
|
|
MQTTClientID: getEnv("INGEST_MQTT_CLIENT_ID", "ingest-service"),
|
|
MQTTTopic: getEnv("INGEST_MQTT_TOPIC", "devices/+/telemetry"),
|
|
|
|
ClickHouseAddr: fmt.Sprintf("%s:%s", getEnv("CLICKHOUSE_HOST", "localhost"), getEnv("CLICKHOUSE_NATIVE_PORT", "9000")),
|
|
ClickHouseDatabase: getEnv("CLICKHOUSE_DB", "telemetry"),
|
|
ClickHouseUsername: getEnv("CLICKHOUSE_USER", "default"),
|
|
ClickHousePassword: getEnv("CLICKHOUSE_PASSWORD", ""),
|
|
|
|
RabbitMQURL: fmt.Sprintf("amqp://%s:%s@%s:%s/",
|
|
getEnv("RABBITMQ_USER", "guest"),
|
|
getEnv("RABBITMQ_PASSWORD", "guest"),
|
|
getEnv("RABBITMQ_HOST", "localhost"),
|
|
getEnv("RABBITMQ_PORT", "5672"),
|
|
),
|
|
}
|
|
|
|
var err error
|
|
if cfg.BatchMaxSize, err = getEnvInt("INGEST_BATCH_MAX_SIZE", 500); err != nil {
|
|
return Config{}, err
|
|
}
|
|
if cfg.BatchFlushInterval, err = getEnvDuration("INGEST_BATCH_FLUSH_INTERVAL", 5*time.Second); err != nil {
|
|
return Config{}, err
|
|
}
|
|
if cfg.BatchFlushTimeout, err = getEnvDuration("INGEST_BATCH_FLUSH_TIMEOUT", 10*time.Second); err != nil {
|
|
return Config{}, err
|
|
}
|
|
|
|
return cfg, nil
|
|
}
|
|
|
|
func getEnv(key, fallback string) string {
|
|
if v := os.Getenv(key); v != "" {
|
|
return v
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func getEnvInt(key string, fallback int) (int, error) {
|
|
v := os.Getenv(key)
|
|
if v == "" {
|
|
return fallback, nil
|
|
}
|
|
n, err := strconv.Atoi(v)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("%s: %w", key, err)
|
|
}
|
|
return n, nil
|
|
}
|
|
|
|
func getEnvDuration(key string, fallback time.Duration) (time.Duration, error) {
|
|
v := os.Getenv(key)
|
|
if v == "" {
|
|
return fallback, nil
|
|
}
|
|
d, err := time.ParseDuration(v)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("%s: %w", key, err)
|
|
}
|
|
return d, nil
|
|
}
|