Misc Updates

cmd/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	_ "go.uber.org/automaxprocs"
 	"os"
 	_ "time/tzdata"
 

configs/local/conf/.env

@@ -2,20 +2,21 @@
 PGBOUNCER_VERSION=1.22.1
 
 ## ****** POSTGRES DB ******
-POSTGRESQL_USER="convoy"
-POSTGRESQL_PASSWORD="pg_password"
-POSTGRESQL_DATABASE="convoy"
-POSTGRESQL_HOST="postgres" #should be your db host address
+POSTGRESQL_USER=convoy
+POSTGRESQL_PASSWORD=pg_password
+POSTGRESQL_DATABASE=convoy
+POSTGRESQL_HOST=postgres #should be your db host address
 POSTGRESQL_OPTIONS="sslmode=disable&connect_timeout=30"
 
 PGBOUNCER_AUTH_TYPE=trust
 PGBOUNCER_USERLIST_FILE=/bitnami/userlists.txt
 PGBOUNCER_DATABASE=${POSTGRESQL_DATABASE}
-PGBOUNCER_AUTH_USER=${POSTGRESQL_USER}
+PGBOUNCER_AUTH_USER=convoy
 PGBOUNCER_POOL_MODE=session
 PGBOUNCER_MAX_CLIENT_CONN=120
 PGBOUNCER_DEFAULT_POOL_SIZE=20
 PGBOUNCER_MAX_DB_CONNECTIONS=250
+PGBOUNCER_IGNORE_STARTUP_PARAMETERS=extra_float_digits
 
 # host should be your db host address
 PGBOUNCER_DSN_0=pg1=host=postgres port=5432 dbname=convoy

configs/local/docker-compose.yml

@@ -6,7 +6,7 @@ services:
         command: ["/start.sh"]
         volumes:
             - ./convoy.json:/convoy.json
-        restart: on-failure
+        restart: unless-stopped
         ports:
             - "5005:5005"
         healthcheck:
@@ -25,17 +25,7 @@ services:
         command: ["./cmd", "worker", "--config", "convoy.json"]
         volumes:
             - ./convoy.json:/convoy.json
-        restart: on-failure
-        depends_on:
-            web:
-                condition: service_healthy
-
-    stream:
-        image: docker.cloudsmith.io/convoy/convoy/frain-dev/convoy:v24.5.1
-        command: ["./cmd", "stream", "--config", "convoy.json"]
-        volumes:
-            - ./convoy.json:/convoy.json
-        restart: on-failure
+        restart: unless-stopped
         depends_on:
             web:
                 condition: service_healthy
@@ -45,7 +35,7 @@ services:
         command: ["./cmd", "ingest", "--config", "convoy.json"]
         volumes:
             - ./convoy.json:/convoy.json
-        restart: on-failure
+        restart: unless-stopped
         depends_on:
             web:
                 condition: service_healthy
@@ -62,25 +52,29 @@ services:
             - ./conf/:/bitnami/pgbouncer/conf/
             - ./conf/userlists.txt:/bitnami/userlists.txt
         ports:
-            - 6432:6432
+            - "6432:6432"
 
     postgres:
-        image: postgres:15.2-alpine
+        image: bitnami/postgresql:latest
         restart: unless-stopped
+        ports:
+            - "5432:5432"
         environment:
-            POSTGRES_DB: convoy
-            POSTGRES_USER: convoy
+            POSTGRESQL_DATABASE: convoy
+            POSTGRESQL_USERNAME: convoy
             POSTGRES_PASSWORD: pg_password
-            PGDATA: /data/postgres
+            POSTGRESQL_MAX_CONNECTIONS: 250
+            POSTGRESQL_POSTGRES_PASSWORD: postgres
+            POSTGRESQL_SHARED_PRELOAD_LIBRARIES: pg_stat_statements
         volumes:
-          - postgres_data:/data/postgres
+          - postgresql_master_data:/bitnami/postgresql
 
     redis_server:
         image: redis:alpine
-        restart: always
+        restart: unless-stopped
         volumes:
           - redis_data:/data
 
 volumes:
-    postgres_data:
+    postgresql_master_data:
     redis_data:

go.mod

@@ -93,9 +93,10 @@ require (
 	github.com/tidwall/pretty v1.2.1 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
-	go.opentelemetry.io/contrib v1.0.0 // indirect
+	go.opentelemetry.io/contrib v1.27.0 // indirect
 	go.opentelemetry.io/otel/metric v1.21.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
+	go.uber.org/automaxprocs v1.5.3 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230822172742-b8732ec3820d // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
 )

go.sum

@@ -714,6 +714,8 @@ go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/contrib v1.0.0 h1:khwDCxdSspjOLmFnvMuSHd/5rPzbTx0+l6aURwtQdfE=
 go.opentelemetry.io/contrib v1.0.0/go.mod h1:EH4yDYeNoaTqn/8yCWQmfNB78VHfGX2Jt2bvnvzBlGM=
+go.opentelemetry.io/contrib v1.27.0 h1:0dNzbHzLqdAT2qoHr9tooz2Iqh+QPyaW01UxNWPZmQ4=
+go.opentelemetry.io/contrib v1.27.0/go.mod h1:Tmhw9grdWtmXy6DxZNpIAudzYJqLeEM2P6QTZQSRwU8=
 go.opentelemetry.io/otel v1.3.0/go.mod h1:PWIKzi6JCp7sM0k9yZ43VX+T345uNbAkDKwHVjb2PTs=
 go.opentelemetry.io/otel v1.21.0 h1:hzLeKBZEL7Okw2mGzZ0cc4k/A7Fta0uoPgaJCr8fsFc=
 go.opentelemetry.io/otel v1.21.0/go.mod h1:QZzNPQPm1zLX4gZK4cMi+71eaorMSGT3A4znnUvNNEo=
@@ -733,6 +735,8 @@ go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqe
 go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I=
 go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM=
 go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
+go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
+go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
 go.uber.org/goleak v0.10.0/go.mod h1:VCZuO8V8mFPlL0F5J5GK1rtHV3DrFcQ1R8ryq7FK0aI=
 go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
 go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=

net/dispatcher.go

@@ -113,7 +113,7 @@ func (d *Dispatcher) do(req *http.Request, res *Response, maxResponseSize int64)
 	trace := &httptrace.ClientTrace{
 		GotConn: func(connInfo httptrace.GotConnInfo) {
 			res.IP = connInfo.Conn.RemoteAddr().String()
-			log.Infof("IP address resolved to: %s", connInfo.Conn.RemoteAddr())
+			log.Debugf("IP address resolved to: %s", connInfo.Conn.RemoteAddr())
 		},
 	}
 

pkg/log/log.go

@@ -102,7 +102,7 @@ func NewLogger(out io.Writer) *Logger {
 		Formatter: &logrus.JSONFormatter{
 			TimestampFormat: "2006-01-02 15:04:05",
 		},
-		Level:        logrus.DebugLevel,
+		Level:        logrus.ErrorLevel,
 		ReportCaller: false,
 	}
 

worker/task/process_event_delivery.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"time"
 
 	"github.com/frain-dev/convoy/internal/pkg/limiter"
@@ -91,7 +90,7 @@ func ProcessEventDelivery(endpointRepo datastore.EndpointRepository, eventDelive
 		if err != nil {
 			log.FromContext(ctx).WithFields(map[string]interface{}{"event_delivery id": data.EventDeliveryID}).
 				WithError(err).
-				Error(fmt.Errorf("too many events to %s, limit of %v reqs/%v has been reached", endpoint.Url, endpoint.RateLimit, time.Duration(endpoint.RateLimitDuration)*time.Second))
+				Debugf("too many events to %s, limit of %v reqs/%v has been reached", endpoint.Url, endpoint.RateLimit, time.Duration(endpoint.RateLimitDuration)*time.Second)
 
 			return &RateLimitError{Err: ErrRateLimit, delay: time.Duration(endpoint.RateLimitDuration) * time.Second}
 		}
@@ -176,26 +175,23 @@ func ProcessEventDelivery(endpointRepo datastore.EndpointRepository, eventDelive
 		})
 
 		if err == nil && statusCode >= 200 && statusCode <= 299 {
-			requestLogger.Infof("%s", eventDelivery.UID)
-			log.Infof("%s sent", eventDelivery.UID)
+			requestLogger.Debugf("%s sent", eventDelivery.UID)
 			attemptStatus = true
-			// e.Sent = true
 
 			eventDelivery.Status = datastore.SuccessEventStatus
 			eventDelivery.Description = ""
 			eventDelivery.Latency = time.Since(eventDelivery.CreatedAt).String()
 		} else {
 			requestLogger.Errorf("%s", eventDelivery.UID)
 			done = false
-			// e.Sent = false
 
 			eventDelivery.Status = datastore.RetryEventStatus
 
 			nextTime := time.Now().Add(delayDuration)
 			eventDelivery.Metadata.NextSendTime = nextTime
 			attempts := eventDelivery.Metadata.NumTrials + 1
 
-			log.FromContext(ctx).Infof("%s next retry time is %s (strategy = %s, delay = %d, attempts = %d/%d)\n", eventDelivery.UID, nextTime.Format(time.ANSIC), eventDelivery.Metadata.Strategy, eventDelivery.Metadata.IntervalSeconds, attempts, eventDelivery.Metadata.RetryLimit)
+			log.FromContext(ctx).Errorf("%s next retry time is %s (strategy = %s, delay = %d, attempts = %d/%d)\n", eventDelivery.UID, nextTime.Format(time.ANSIC), eventDelivery.Metadata.Strategy, eventDelivery.Metadata.IntervalSeconds, attempts, eventDelivery.Metadata.RetryLimit)
 		}
 
 		// Request failed but statusCode is 200 <= x <= 299
@@ -221,7 +217,7 @@ func ProcessEventDelivery(endpointRepo datastore.EndpointRepository, eventDelive
 			endpointStatus := datastore.InactiveEndpointStatus
 			err := endpointRepo.UpdateEndpointStatus(ctx, project.UID, endpoint.UID, endpointStatus)
 			if err != nil {
-				log.FromContext(ctx).Info("Failed to reactivate endpoint after successful retry")
+				log.FromContext(ctx).Errorf("Failed to reactivate endpoint after successful retry")
 			}
 		}
 

worker/task/search_tokenizer.go

@@ -53,9 +53,9 @@ func GeneralTokenizerHandler(projectRepository datastore.ProjectRepository, even
 				log.WithError(err).Errorf("failed to tokenize events for project with id %s", p.Id)
 				continue
 			}
-			log.Printf("done tokenizing events for %+v with %v events\n", p.Id, p.EventsCount)
+			log.Debugf("done tokenizing events for %+v with %v events", p.Id, p.EventsCount)
 		}
-		log.Println("done tokenizing events in the interval")
+		log.Debugf("done tokenizing events in the interval")
 
 		return nil
 	}
@@ -74,7 +74,7 @@ func TokenizerHandler(eventRepo datastore.EventRepository, jobRepo datastore.Job
 		if err != nil {
 			return err
 		}
-		log.Printf("done tokenizing events in the last %d hours for project with id %s", params.Interval, params.ProjectID)
+		log.Debugf("done tokenizing events in the last %d hours for project with id %s", params.Interval, params.ProjectID)
 
 		return nil
 	}