mErge branch 'master' of github.com:anodot/anodot-remote-write

e2e/docker-compose.yaml

@@ -36,6 +36,7 @@ services:
       ANODOT_METRICS_PER_REQUEST_SIZE: 4
       ANODOT_LOG_LEVEL: "5"
       ANODOT_API_TOKEN: "123456"
+      ANODOT_ACCESS_KEY: "123456"
       ANODOT_RELABEL_CONFIG_PATH: "/mnt/relabel.yml"
       ANODOT_HTTP_DEBUG_ENABLED: "true"
       ANODOT_PUSH_METRICS_ENABLED: "true"

go.mod

@@ -1,10 +1,10 @@
 module github.com/anodot/anodot-remote-write
 
-go 1.12
+go 1.13
 
 require (
 	github.com/alecthomas/units v0.0.0-20210208195552-ff826a37aa15 // indirect
-	github.com/anodot/anodot-common v0.0.8
+	github.com/anodot/anodot-common v0.0.10
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2
 	github.com/golang/snappy v0.0.3

main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"flag"
 	"fmt"
+	"github.com/anodot/anodot-common/pkg/metrics3"
 	"net/http"
 	_ "net/http/pprof"
 	"net/url"
@@ -195,6 +196,23 @@ func main() {
 		reporter.Report()
 	}
 
+	ifSendToBC := defaultIfBlank(os.Getenv("ANODOT_SEND_TO_BC"), "true")
+	sendToBCPeriod, err := strconv.Atoi(defaultIfBlank(os.Getenv("ANODOT_SEND_TO_BC_PERIOD_SEC"), "60"))
+	if err != nil {
+		log.Fatalf("Could not parse ANODOT_SEND_TO_BC_PERIOD_SEC: %v", err)
+	}
+	if ifSendToBC != "false" {
+		accessKey := os.Getenv("ANODOT_ACCESS_KEY")
+		if len(strings.TrimSpace(accessKey)) == 0 {
+			log.Fatalf("ANODOT_ACCESS_KEY is not specified")
+		}
+		client, err := metrics3.NewAnodot30Client(*primaryUrl, &accessKey, &token, nil)
+		if err != nil {
+			log.Fatalf("failed to create anodot30 client: %v", err)
+		}
+		anodotPrometheus.SendAgentStatusToBC(client, sendToBCPeriod)
+	}
+
 	s.InitHttp(allWorkers)
 }
 

pkg/prometheus/bc.go

@@ -0,0 +1,45 @@
+package prometheus
+
+import (
+	"github.com/anodot/anodot-common/pkg/metrics3"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	log "k8s.io/klog/v2"
+	"time"
+)
+
+func NewPipeline(startTime metrics3.AnodotTimestamp) metrics3.Pipeline {
+	pipeline := metrics3.Pipeline{}
+	pipeline.Id = "prometheus"
+	pipeline.Created = startTime
+	pipeline.Updated = startTime
+	pipeline.Status = "RUNNING"
+
+	s := metrics3.Source{}
+	s.Name = "prometheus"
+	s.Type = "prometheus"
+	pipeline.Source = s
+
+	return pipeline
+}
+
+var sendStatusToBCErrors = promauto.NewCounter(prometheus.CounterOpts{
+	Name: "anodot_send_status_to_bc_errors_total",
+	Help: "Total number of errors occurred while sending status to BC",
+})
+
+func SendAgentStatusToBC(client *metrics3.Anodot30Client, sendToBCPeriod int) {
+	startTime := metrics3.AnodotTimestamp{Time: time.Now()}
+	go func() {
+		ticker := time.NewTicker(time.Duration(sendToBCPeriod) * time.Second)
+		defer ticker.Stop()
+
+		for range ticker.C {
+			_, err := client.SendToBC(NewPipeline(startTime))
+			if err != nil {
+				sendStatusToBCErrors.Inc()
+				log.Errorf("Failed to send status to BC %v", err)
+			}
+		}
+	}()
+}