Restart vector if it stops receiving logs from sources

Dockerfile

@@ -1,5 +1,14 @@
-FROM timberio/vector:0.29.1-debian
+FROM golang:1.21 as builder
+
+WORKDIR /app
+COPY vector-monitor.go .
+RUN go mod init vector-monitor
+RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o vector-monitor .
+
+FROM timberio/vector:0.33.0-debian
 COPY vector-configs /etc/vector/
 COPY ./start-fly-log-transporter.sh .
+COPY --from=builder /app/vector-monitor /usr/local/bin/
+
 CMD ["bash", "start-fly-log-transporter.sh"]
 ENTRYPOINT []

fly.toml

@@ -1,4 +1,7 @@
 app = "fly-log-shipper-example"
+primary_region = "fra"
+[env]
+  ORG = "myorg"
 
 [metrics]
   port = 9598

start-fly-log-transporter.sh

@@ -8,4 +8,20 @@ filter /etc/vector/sinks/*.toml 2>&-
 echo 'Configured sinks:'
 find /etc/vector/sinks -type f -exec basename -s '.toml' {} \;
 
-exec vector -c /etc/vector/vector.toml -C /etc/vector/sinks
+
+vector -c /etc/vector/vector.toml -C /etc/vector/sinks &
+/usr/local/bin/vector-monitor &
+
+VECTOR_PID=$!
+MONITOR_PID=$!
+
+cleanup() {
+    echo "Shutting down services..."
+    kill $VECTOR_PID
+    kill $MONITOR_PID
+    exit 0
+}
+
+trap cleanup SIGINT SIGTERM
+
+wait

vector-configs/vector.toml

@@ -2,9 +2,6 @@
   enabled = true
   address = "0.0.0.0:8686"
 
-[sources.fly_log_metrics]
-  type = "internal_metrics"
-
 [sources.nats]
   type = "nats"
   url = "nats://[fdaa::3]:4223"
@@ -22,12 +19,6 @@
   . = parse_json!(.message)
   '''
 
-[sinks.fly_log_metrics_prometheus]
-  type = "prometheus_exporter" # required
-  inputs = ["fly_log_metrics"] # required
-  address = "0.0.0.0:9598" # required
-  default_namespace = "fly-logs" # optional, no default
-
 [sinks.blackhole]
   type = "blackhole"
   inputs = ["log_json"]

vector-monitor.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"net/http"
+	"os/exec"
+	"os"
+	"strconv"
+	"time"
+)
+
+const (
+	vectorAPIURL      = "http://localhost:8686/graphql"
+	defaultInactivityTimeout = 10
+	vectorExecutable  = "vector" // path to the Vector executable if not in $PATH
+)
+
+type GraphqlRequest struct {
+	Query string `json:"query"`
+}
+
+type GraphqlResponse struct {
+	Data struct {
+		Metrics struct {
+			ProcessedEventsTotal float64 `json:"processedEventsTotal"`
+		} `json:"metrics"`
+	} `json:"data"`
+}
+
+func getInactivityTimeout() time.Duration {
+	timeout := defaultInactivityTimeout
+
+	// Read the environment variable
+	if value, exists := os.LookupEnv("INACTIVITY_TIMEOUT"); exists {
+		if intValue, err := strconv.Atoi(value); err == nil && intValue > 0 {
+			timeout = intValue
+		} else {
+			log.Printf("Warning: Invalid INACTIVITY_TIMEOUT value, using default %d seconds", defaultInactivityTimeout)
+		}
+	}
+
+	return time.Duration(timeout) * time.Second
+}
+
+func getProcessedEvents() (float64, error) {
+	payload := GraphqlRequest{
+		Query: `
+		{
+			metrics {
+				processedEventsTotal
+			}
+		}`,
+	}
+
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return 0, err
+	}
+
+	resp, err := http.Post(vectorAPIURL, "application/json", bytes.NewBuffer(body))
+	if err != nil {
+		return 0, err
+	}
+	defer resp.Body.Close()
+
+	var result GraphqlResponse
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return 0, err
+	}
+
+	return result.Data.Metrics.ProcessedEventsTotal, nil
+}
+
+func restartVector() {
+	cmd := exec.Command(vectorExecutable)
+	if err := cmd.Start(); err != nil {
+		log.Fatalf("Error restarting Vector: %v", err)
+	}
+	log.Println("Vector restarted.")
+}
+
+func main() {
+	var previousCount float64
+	lastActivityTime := time.Now()
+	var inactivityTimeout = getInactivityTimeout()
+	log.Println("Vector monitor started with threshold of %s seconds of inactivity.", inactivityTimeout)
+	for {
+		time.Sleep(inactivityTimeout * time.Second)
+
+		currentCount, err := getProcessedEvents()
+		if err != nil {
+			log.Printf("Error retrieving event count: %v", err)
+			continue
+		}
+
+		if currentCount != previousCount {
+			lastActivityTime = time.Now()
+		} else if time.Since(lastActivityTime) > inactivityTimeout {
+			log.Println("No new logs processed for the specified duration. Restarting Vector.")
+			restartVector()
+			lastActivityTime = time.Now() // Reset the timer after restart
+		}
+
+		previousCount = currentCount
+	}
+}