This repository has been archived by the owner on Aug 25, 2022. It is now read-only.
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
PLAT-301 Migrated to viper for config files and cobra for app flags
- Loading branch information
Showing
13 changed files
with
293 additions
and
318 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,21 +1,21 @@ | ||
| # Available options are: "debug", "info", "warning", "error", "fatal" and "panic" | ||
| log_level: "info" | ||
| rabbit_dsn: "amqp://user:password@rmq" | ||
| storage_dsn: "redis://redis.local/?key=storage:key" | ||
| logLevel: "info" | ||
| rabbitDSN: "amqp://user:password@rmq" | ||
| storageDSN: "redis://redis.local/?key=storage:key" | ||
| kafka: | ||
| brokers: | ||
| - "192.0.0.1:9092" | ||
| - "192.0.0.2:9092" | ||
| # The total number of times to retry sending a message. | ||
| # Should be similar to the `message.send.max.retries` setting of the JVM producer. | ||
| max_retry: 5 | ||
| pipes_config: "/etc/kandalf/conf/pipes.yml" | ||
| maxRetry: 5 | ||
| pipesConfig: "/etc/kandalf/conf/pipes.yml" | ||
| stats: | ||
| dsn: "statsd.local:8125" | ||
| prefix: "kandalf" | ||
| worker: | ||
| cycle_timeout: "2s" | ||
| cache_size: 10 | ||
| cache_flush_timeout: "5s" | ||
| storage_read_timeout: "10s" | ||
| storage_max_errors: 10 | ||
| cycleTimeout: "2s" | ||
| cacheSize: 10 | ||
| cacheFlushTimeout: "5s" | ||
| storageReadTimeout: "10s" | ||
| storageMaxErrors: 10 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,16 +1,14 @@ | ||
| - | ||
| pipes: | ||
| # Message from that RabbitMQ exchange | ||
| rabbitmq_exchange_name: "customers" | ||
| - rabbitExchangeName: "customers" | ||
| # With that routing key | ||
| rabbitmq_routing_key: "order.created" | ||
| rabbitRoutingKey: "order.created" | ||
| # Will be placed to that kafka topic | ||
| kafka_topic: "new-orders" | ||
| kafkaTopic: "new-orders" | ||
| # The queue name can be whatever you want, just keep it unique within pipes. | ||
| # If you launch multiple kandalf instances they all will consume messages from that queue. | ||
| rabbitmq_queue_name: "kandalf-customers-order.created" | ||
| rabbitQueueName: "kandalf-customers-order.created" | ||
|
|
||
| - | ||
| kafka_topic: "loyalty" | ||
| rabbitmq_exchange_name: "customers" | ||
| rabbitmq_routing_key: "badge.received" | ||
| rabbitmq_queue_name: "kandalf-customers-badge.received" | ||
| - kafkaTopic: "loyalty" | ||
| rabbitExchangeName: "customers" | ||
| rabbitRoutingKey: "badge.received" | ||
| rabbitQueueName: "kandalf-customers-badge.received" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "net/url" | ||
| "strings" | ||
|
|
||
| log "github.com/Sirupsen/logrus" | ||
| "github.com/hellofresh/kandalf/pkg/amqp" | ||
| "github.com/hellofresh/kandalf/pkg/config" | ||
| "github.com/hellofresh/kandalf/pkg/kafka" | ||
| "github.com/hellofresh/kandalf/pkg/storage" | ||
| "github.com/hellofresh/kandalf/pkg/workers" | ||
| "github.com/hellofresh/stats-go" | ||
| "github.com/spf13/cobra" | ||
| ) | ||
|
|
||
| // RunApp is main application bootstrap and runner | ||
| func RunApp(cmd *cobra.Command, args []string) { | ||
| printVersion, err := cmd.Flags().GetBool(flagVersion) | ||
| failOnError(err, "Failed to read version flag") | ||
| if printVersion { | ||
| cmd.Println(cmd.Short) | ||
| return | ||
| } | ||
|
|
||
| log.WithField("version", version).Info("Kandalf starting...") | ||
|
|
||
| globalConfig, err := config.Load(configPath) | ||
| failOnError(err, "Failed to load application configuration") | ||
|
|
||
| level, err := log.ParseLevel(strings.ToLower(globalConfig.LogLevel)) | ||
| failOnError(err, "Failed to get log level") | ||
| log.SetLevel(level) | ||
|
|
||
| pipesList, err := config.LoadPipesFromFile(globalConfig.Kafka.PipesConfig) | ||
| failOnError(err, "Failed to load pipes config") | ||
|
|
||
| statsClient := stats.NewStatsdStatsClient(globalConfig.Stats.DSN, globalConfig.Stats.Prefix) | ||
| defer func() { | ||
| if err := statsClient.Close(); err != nil { | ||
| log.WithError(err).Error("Got error on closing stats client") | ||
| } | ||
| }() | ||
|
|
||
| storageURL, err := url.Parse(globalConfig.StorageDSN) | ||
| failOnError(err, "Failed to parse Storage DSN") | ||
|
|
||
| persistentStorage, err := storage.NewPersistentStorage(storageURL) | ||
| failOnError(err, "Failed to establish Redis connection") | ||
| // Do not close storage here as it is required in Worker close to store unhandled messages | ||
|
|
||
| producer, err := kafka.NewProducer(globalConfig.Kafka, statsClient) | ||
| failOnError(err, "Failed to establish Kafka connection") | ||
| defer func() { | ||
| if err := producer.Close(); err != nil { | ||
| log.WithError(err).Error("Got error on closing kafka producer") | ||
| } | ||
| }() | ||
|
|
||
| worker, err := workers.NewBridgeWorker(globalConfig.Worker, persistentStorage, producer, statsClient) | ||
| defer func() { | ||
| if err := worker.Close(); err != nil { | ||
| log.WithError(err).Error("Got error on closing persistent storage") | ||
| } | ||
| }() | ||
|
|
||
| queuesHandler := amqp.NewQueuesHandler(pipesList, worker.MessageHandler, statsClient) | ||
| amqpConnection, err := amqp.NewConnection(globalConfig.RabbitDSN, queuesHandler) | ||
| failOnError(err, "Failed to establish initial connection to AMQP") | ||
| defer func() { | ||
| if err := amqpConnection.Close(); err != nil { | ||
| log.WithError(err).Error("Got error on closing AMQP connection") | ||
| } | ||
| }() | ||
|
|
||
| forever := make(chan bool) | ||
|
|
||
| worker.Go(forever) | ||
|
|
||
| log.Infof("[*] Waiting for users. To exit press CTRL+C") | ||
| <-forever | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.