refactor: docker-compose files (#104)

ormushq · Jul 17, 2024 · 5abba61 · 5abba61
1 parent 85224aa
commit 5abba61
Show file tree

Hide file tree

Showing 51 changed files with 279 additions and 119 deletions.
diff --git a/.gitignore b/.gitignore
@@ -32,3 +32,5 @@ destination/worker_log.json
 temp
 .env
 ormus.iml
+
+vendor
diff --git a/adapter/etcd/adapter.go b/adapter/etcd/adapter.go
@@ -3,14 +3,14 @@ package etcd
 import (
 	"context"
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log/slog"
 	"time"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Config struct {

diff --git a/adapter/otela/metric.go b/adapter/otela/metric.go
@@ -3,6 +3,7 @@ package otela
 import (
 	"context"
 	"fmt"
+
 	"github.com/ormushq/ormus/logger"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -11,7 +12,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 	sdkMetric "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/sdk/resource"
-	"go.opentelemetry.io/otel/semconv/v1.25.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -24,7 +25,6 @@ func (opr *otelProvider) newMetricExporter(ctx context.Context) (sdkMetric.Expor
 	default:
 		panic("unsupported")
 	}
-
 }
 
 func (opr *otelProvider) newMetricProvider(exp sdkMetric.Exporter) *sdkMetric.MeterProvider {

diff --git a/adapter/otela/otel.go b/adapter/otela/otel.go
@@ -20,6 +20,7 @@ var op otelProvider
 
 func Configure(wg *sync.WaitGroup, done <-chan bool, cfg Config) error {
 	if cfg.EnableMetricExpose && !op.isConfigure {
+		fmt.Println(cfg)
 		serveMetric(cfg, wg, done)
 	}
 

diff --git a/adapter/otela/trace.go b/adapter/otela/trace.go
@@ -3,15 +3,15 @@ package otela
 import (
 	"context"
 	"fmt"
-	"go.opentelemetry.io/otel/propagation"
 
 	"github.com/ormushq/ormus/logger"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
 	"go.opentelemetry.io/otel/exporters/stdout/stdouttrace"
+	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
-	"go.opentelemetry.io/otel/semconv/v1.25.0"
+	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -24,7 +24,6 @@ func (opr *otelProvider) newTraceExporter(ctx context.Context) (sdktrace.SpanExp
 	default:
 		panic("unsupported")
 	}
-
 }
 
 func (opr *otelProvider) newTraceProvider(exp sdktrace.SpanExporter) *sdktrace.TracerProvider {

diff --git a/cmd/destination/delivery_workers/webhook_delivery_worker.go b/cmd/destination/delivery_workers/webhook_delivery_worker.go
@@ -2,11 +2,6 @@ package main
 
 import (
 	"context"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/adapter/redis"
-	"github.com/ormushq/ormus/destination/taskservice/adapter/idempotency/redistaskidempotency"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log"
 	"log/slog"
 	"os"
@@ -15,17 +10,22 @@ import (
 	"time"
 
 	"github.com/ormushq/ormus/adapter/etcd"
+	"github.com/ormushq/ormus/adapter/otela"
+	"github.com/ormushq/ormus/adapter/redis"
 	"github.com/ormushq/ormus/config"
 	"github.com/ormushq/ormus/destination/dconfig"
 	"github.com/ormushq/ormus/destination/taskdelivery"
 	"github.com/ormushq/ormus/destination/taskdelivery/adapters/fakedeliveryhandler"
 	"github.com/ormushq/ormus/destination/taskmanager/adapter/rabbitmqchanneltaskmanager"
 	"github.com/ormushq/ormus/destination/taskservice"
+	"github.com/ormushq/ormus/destination/taskservice/adapter/idempotency/redistaskidempotency"
 	"github.com/ormushq/ormus/destination/taskservice/adapter/repository/inmemorytaskrepo"
 	"github.com/ormushq/ormus/destination/worker"
 	"github.com/ormushq/ormus/logger"
 	"github.com/ormushq/ormus/pkg/channel"
 	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 const waitingAfterShutdownInSeconds = 2

diff --git a/cmd/destination/faker/fake_processed_event_producer.go b/cmd/destination/faker/fake_processed_event_producer.go
@@ -5,23 +5,23 @@ import (
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/pkg/metricname"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log"
 	"log/slog"
 	"math/big"
 	"os"
 	"sync"
 	"time"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/config"
 	"github.com/ormushq/ormus/event"
 	"github.com/ormushq/ormus/logger"
 	"github.com/ormushq/ormus/manager/entity"
+	"github.com/ormushq/ormus/pkg/metricname"
 	amqp "github.com/rabbitmq/amqp091-go"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 const timeoutSeconds = 5

diff --git a/cmd/destination/main.go b/cmd/destination/main.go
@@ -2,16 +2,14 @@ package main
 
 import (
 	"context"
-	"github.com/ormushq/ormus/adapter/otela"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log"
 	"log/slog"
 	"os"
 	"os/signal"
 	"sync"
 	"time"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/config"
 	"github.com/ormushq/ormus/destination/dconfig"
 	"github.com/ormushq/ormus/destination/processedevent/adapter/rabbitmqconsumer"
@@ -21,6 +19,8 @@ import (
 	"github.com/ormushq/ormus/manager/entity"
 	"github.com/ormushq/ormus/pkg/channel"
 	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 const waitingAfterShutdownInSeconds = 1

diff --git a/cmd/example/otel/main.go b/cmd/example/otel/main.go
@@ -3,12 +3,12 @@ package main
 import (
 	"context"
 	"fmt"
-	"go.opentelemetry.io/otel/metric"
 	"sync"
 	"time"
 
 	"github.com/ormushq/ormus/adapter/otela"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -45,6 +45,7 @@ func main() {
 	close(done)
 	wg.Wait()
 }
+
 func startService1(c chan<- context.Context) {
 	tracer := otela.NewTracer("test-tracer")
 

diff --git a/cmd/example/otel/rabbitmq/consumer/main.go b/cmd/example/otel/rabbitmq/consumer/main.go
@@ -3,13 +3,14 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"os"
+	"os/signal"
+	"sync"
+
 	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/destination/dconfig"
 	"github.com/ormushq/ormus/pkg/channel"
 	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
-	"os"
-	"os/signal"
-	"sync"
 )
 
 type MyMessage struct {
@@ -61,7 +62,7 @@ func main() {
 			select {
 			case msg := <-outputChannel:
 				func() {
-					fmt.Printf("recived message : %s\n", msg.Body)
+					fmt.Printf("received message : %s\n", msg.Body)
 					var decode MyMessage
 					err = json.Unmarshal(msg.Body, &decode)
 					if err != nil {

diff --git a/cmd/example/otel/rabbitmq/publisher/main.go b/cmd/example/otel/rabbitmq/publisher/main.go
@@ -4,14 +4,15 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/destination/dconfig"
-	"github.com/ormushq/ormus/pkg/channel"
-	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
 	"os"
 	"os/signal"
 	"sync"
 	"time"
+
+	"github.com/ormushq/ormus/adapter/otela"
+	"github.com/ormushq/ormus/destination/dconfig"
+	"github.com/ormushq/ormus/pkg/channel"
+	rbbitmqchannel "github.com/ormushq/ormus/pkg/channel/adapter/rabbitmq"
 )
 
 type MyMessage struct {

diff --git a/config.yml b/config.yml
@@ -10,31 +10,31 @@ scylladb:
   keyspace: ormus
 redis:
   port: 6379
-  host: localhost
+  host: redis
   db: 0
   password: ""
 etcd:
   port: 2379
-  host: localhost
+  host: etcd
   dial_timeout: 5
 destination:
   debug_mode: true
   consumer_topic: "pe.#"    # pe stands for processed event. and # substitute for zero or more words.
   rabbitmq_consumer_connection:
     user: guest
     password: guest
-    host: localhost
+    host: rabbitmq
     port: 5672
     vhost:
     reconnect_second: 5
   rabbitmq_task_manager_connection:
     user: guest
     password: guest
-    host: localhost
+    host: rabbitmq
     port: 5672
   redis_task_idempotency:
     port: 6379
-    host: localhost
+    host: redis
     db: 0
     password: ""
     prefix: "destination-task-status::"

diff --git a/destination/processedevent/adapter/rabbitmqconsumer/adapter.go b/destination/processedevent/adapter/rabbitmqconsumer/adapter.go
@@ -2,22 +2,20 @@ package rabbitmqconsumer
 
 import (
 	"context"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/pkg/metricname"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-
-	//"context"
 	"fmt"
 	"log"
 	"log/slog"
 	"sync"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/destination/dconfig"
 	"github.com/ormushq/ormus/destination/entity/taskentity"
 	"github.com/ormushq/ormus/event"
+	"github.com/ormushq/ormus/pkg/metricname"
 	amqp "github.com/rabbitmq/amqp091-go"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Consumer struct {
@@ -48,6 +46,7 @@ func (c *Consumer) Consume(done <-chan bool, wg *sync.WaitGroup) (<-chan event.P
 		defer close(events)
 		_, span := tracer.Start(c.ctx, "rabbitmqconsumer@consume")
 
+		fmt.Println(c.connectionConfig)
 		conn, err := amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d/", c.connectionConfig.User, c.connectionConfig.Password, c.connectionConfig.Host, c.connectionConfig.Port))
 		failOnError(err, "Failed to connect to RabbitMQ")
 		defer func(conn *amqp.Connection) {

diff --git a/destination/taskcoordinator/adapter/dtcoordinator/adapter.go b/destination/taskcoordinator/adapter/dtcoordinator/adapter.go
@@ -2,17 +2,17 @@ package dtcoordinator
 
 import (
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/pkg/metricname"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log/slog"
 	"sync"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/destination/taskmanager"
 	"github.com/ormushq/ormus/event"
 	"github.com/ormushq/ormus/manager/entity"
+	"github.com/ormushq/ormus/pkg/metricname"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type TaskPublisherMap map[entity.DestinationType]taskmanager.Publisher

diff --git a/destination/taskdelivery/adapters/fakedeliveryhandler/handler.go b/destination/taskdelivery/adapters/fakedeliveryhandler/handler.go
@@ -2,14 +2,14 @@ package fakedeliveryhandler
 
 import (
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"log/slog"
 	"time"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/destination/entity/taskentity"
 	"github.com/ormushq/ormus/destination/taskdelivery/param"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type FakeHandler struct{}

diff --git a/destination/taskmanager/adapter/inmemorytaskmanager/worker.go b/destination/taskmanager/adapter/inmemorytaskmanager/worker.go
@@ -3,15 +3,15 @@ package inmemorytaskmanager
 import (
 	"sync"
 
-	"github.com/ormushq/ormus/destination/taskmanager"
+	"github.com/ormushq/ormus/destination/worker"
 )
 
 type Worker struct {
 	TaskManager *TaskManager
-	Handler     taskmanager.TaskHandler
+	Handler     worker.TaskHandler
 }
 
-func NewWorker(tm *TaskManager, h taskmanager.TaskHandler) *Worker {
+func NewWorker(tm *TaskManager, h worker.TaskHandler) *Worker {
 	return &Worker{
 		TaskManager: tm,
 		Handler:     h,

diff --git a/destination/taskmanager/adapter/rabbitmqchanneltaskmanager/task_consumer.go b/destination/taskmanager/adapter/rabbitmqchanneltaskmanager/task_consumer.go
@@ -3,17 +3,17 @@ package rabbitmqchanneltaskmanager
 import (
 	"context"
 	"fmt"
-	"github.com/ormushq/ormus/adapter/otela"
-	"github.com/ormushq/ormus/pkg/metricname"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
 	"sync"
 
+	"github.com/ormushq/ormus/adapter/otela"
 	"github.com/ormushq/ormus/destination/entity/taskentity"
 	"github.com/ormushq/ormus/event"
 	"github.com/ormushq/ormus/logger"
 	"github.com/ormushq/ormus/pkg/channel"
+	"github.com/ormushq/ormus/pkg/metricname"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 type Consumer struct {