Add support for InfluxDB 2 and InfluxDB 3 Core

cmd/go.mod

@@ -119,6 +119,13 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 
-require github.com/euank/go-kmsg-parser v2.0.0+incompatible // indirect
+require github.com/influxdata/influxdb-client-go/v2 v2.14.0
+
+require (
+	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
+	github.com/euank/go-kmsg-parser v2.0.0+incompatible // indirect
+	github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 // indirect
+	github.com/oapi-codegen/runtime v1.0.0 // indirect
+)
 
 replace github.com/google/cadvisor/lib => ../lib

cmd/go.sum

@@ -6,6 +6,7 @@ cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdB
 cloud.google.com/go/compute/metadata v0.9.0/go.mod h1:E0bWwX5wTnLPedCKqk3pJmVgCBSM6qQI1yTBdEb3C10=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
 github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
+github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
 github.com/SeanDolphin/bqschema v1.0.0 h1:iCYFd5Qsw6caM2k5/SsITSL9+3kQCr+oz6pnNjWTq90=
 github.com/SeanDolphin/bqschema v1.0.0/go.mod h1:TYInVncsPIZH7kybQoIUNJ4pFX1cUc8LoP9RSOxIs6c=
 github.com/Shopify/sarama v1.38.1 h1:lqqPUPQZ7zPqYlWpTh+LQ9bhYNu2xJL6k1SJN4WVe2A=
@@ -18,6 +19,8 @@ github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNg
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/apache/arrow/go/v7 v7.0.1 h1:WpCfq+AQxvXaI6/KplHE27MPMFx5av0o5NbPCTAGfy4=
 github.com/apache/arrow/go/v7 v7.0.1/go.mod h1:JxDpochJbCVxqbX4G8i1jRqMrnTCQdf8pTccAfLD8Es=
+github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
+github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=
 github.com/aws/aws-sdk-go-v2 v1.36.3 h1:mJoei2CxPutQVxaATCzDUjcZEjVRdpsiiXi2o38yqWM=
 github.com/aws/aws-sdk-go-v2 v1.36.3/go.mod h1:LLXuLpgzEbD766Z5ECcRmi8AzSwfZItDtmABVkRLGzg=
 github.com/aws/aws-sdk-go-v2/config v1.29.14 h1:f+eEi/2cKCg9pqKBoAIwRGzVb70MRKqWX4dg1BDcSJM=
@@ -50,6 +53,7 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
 github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
+github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/containerd/containerd/api v1.10.0 h1:5n0oHYVBwN4VhoX9fFykCV9dF1/BvAXeg2F8W6UYq1o=
@@ -147,8 +151,12 @@ github.com/influxdata/flux v0.196.1 h1:RZypfrrAZeIixD/2As+qoyrwfs6Gx7VcnTRNGNjBf
 github.com/influxdata/flux v0.196.1/go.mod h1:+Y4mBygx6q98onpdKJd6vJPrTNjHriQhwh/gM+3IvUQ=
 github.com/influxdata/influxdb v1.12.0 h1:hVFeHtEUh/MkI6YCmB1TAXlPgV4ZNh/V5FP6z7ywRLo=
 github.com/influxdata/influxdb v1.12.0/go.mod h1:11RjLuBNkuWaJQFViRF/rpNzICfU6X0nuO003yeleKY=
+github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjwJdUHnwvfjMF71M1iI4=
+github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI=
 github.com/influxdata/influxql v1.4.1 h1:UB+TMc9cB6mDdkPmH/5sBU8FQ+ZCWRX2JPcPDIFrLcs=
 github.com/influxdata/influxql v1.4.1/go.mod h1:VqxAKyQz5p8GzgGsxWalCWYGxEqw6kvJo2IickMQiQk=
+github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839 h1:W9WBk7wlPfJLvMCdtV4zPulc4uCPrlywQOmbFOhgQNU=
+github.com/influxdata/line-protocol v0.0.0-20200327222509-2487e7298839/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
 github.com/influxdb/influxdb v1.7.9 h1:KMBwwvyJyBppIwrg5t0662p+Yei/ucnIkqUl8txiQdQ=
 github.com/influxdb/influxdb v1.7.9/go.mod h1:GpjLgHRqWhDGlPAg7+Rj6NAYuzPojBM8XLG5Ouvvq+Q=
 github.com/jcmturner/aescts/v2 v2.0.0 h1:9YKLH6ey7H4eDBXW8khjYslgyqG2xZikXP0EQFKrle8=
@@ -163,6 +171,7 @@ github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh6
 github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs=
 github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY=
 github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc=
+github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=
 github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -188,6 +197,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8m
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
+github.com/oapi-codegen/runtime v1.0.0 h1:P4rqFX5fMFWqRzY9M/3YF9+aPSPPB06IzP2P7oOxrWo=
+github.com/oapi-codegen/runtime v1.0.0/go.mod h1:LmCUMQuPB4M/nLXilQXhHw+BLZdDb18B34OO356yJ/A=
 github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
@@ -230,11 +241,13 @@ github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/spkg/bom v0.0.0-20160624110644-59b7046e48ad/go.mod h1:qLr4V1qq6nMqFKkMo8ZTx3f+BZEkzsRUY10Xsm2mwU0=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=

cmd/internal/storage/influxdb/influxdb.go

@@ -15,6 +15,7 @@
 package influxdb
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"net/url"
@@ -26,19 +27,30 @@ import (
 	"github.com/google/cadvisor/lib/storage"
 	"github.com/google/cadvisor/lib/version"
 
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	api "github.com/influxdata/influxdb-client-go/v2/api"
+	"github.com/influxdata/influxdb-client-go/v2/api/write"
 	influxdb "github.com/influxdb/influxdb/client"
 )
 
 func init() {
 	storage.RegisterStorageDriver("influxdb", new)
 }
 
-var argDbRetentionPolicy = flag.String("storage_driver_influxdb_retention_policy", "", "retention policy")
+var (
+	argDbRetentionPolicy = flag.String("storage_driver_influxdb_retention_policy", "", "retention policy")
+	argDbAuthToken       = flag.String("storage_driver_influxdb_auth_token", "", "InfluxDB API token. When set, stats are written through the InfluxDB 2.x API, which is also implemented by InfluxDB 3 Core and InfluxDB Cloud, using token authentication instead of the InfluxDB 1.x username/password API")
+	argDbOrg             = flag.String("storage_driver_influxdb_org", "", "InfluxDB organization to write to. Required by InfluxDB 2.x, ignored by InfluxDB 3 Core")
+	argDbBucket          = flag.String("storage_driver_influxdb_bucket", "", "InfluxDB 2.x bucket or InfluxDB 3 Core database to write to. Defaults to the -storage_driver_db value")
+)
 
 type influxdbStorage struct {
 	client          *influxdb.Client
+	v2Client        influxdb2.Client
+	writeAPI        api.WriteAPIBlocking
 	machineName     string
 	database        string
+	bucket          string
 	retentionPolicy string
 	bufferDuration  time.Duration
 	lastWrite       time.Time
@@ -120,6 +132,9 @@ func new() (storage.StorageDriver, error) {
 		*argDbRetentionPolicy,
 		*storage.ArgDbUsername,
 		*storage.ArgDbPassword,
+		*argDbAuthToken,
+		*argDbOrg,
+		*argDbBucket,
 		*storage.ArgDbHost,
 		*storage.ArgDbIsSecure,
 		*storage.ArgDbBufferDuration,
@@ -208,31 +223,35 @@ func (s *influxdbStorage) containerStatsToPoints(
 	stats *info.ContainerStats,
 ) (points []*influxdb.Point) {
 	// CPU usage: Total usage in nanoseconds
-	points = append(points, makePoint(serCPUUsageTotal, stats.Cpu.Usage.Total))
+	if stats.Cpu != nil {
+		points = append(points, makePoint(serCPUUsageTotal, stats.Cpu.Usage.Total))
 
-	// CPU usage: Time spend in system space (in nanoseconds)
-	points = append(points, makePoint(serCPUUsageSystem, stats.Cpu.Usage.System))
+		// CPU usage: Time spend in system space (in nanoseconds)
+		points = append(points, makePoint(serCPUUsageSystem, stats.Cpu.Usage.System))
 
-	// CPU usage: Time spent in user space (in nanoseconds)
-	points = append(points, makePoint(serCPUUsageUser, stats.Cpu.Usage.User))
+		// CPU usage: Time spent in user space (in nanoseconds)
+		points = append(points, makePoint(serCPUUsageUser, stats.Cpu.Usage.User))
 
-	// CPU usage per CPU
-	for i := 0; i < len(stats.Cpu.Usage.PerCpu); i++ {
-		point := makePoint(serCPUUsagePerCPU, stats.Cpu.Usage.PerCpu[i])
-		tags := map[string]string{"instance": fmt.Sprintf("%v", i)}
-		addTagsToPoint(point, tags)
+		// CPU usage per CPU
+		for i := 0; i < len(stats.Cpu.Usage.PerCpu); i++ {
+			point := makePoint(serCPUUsagePerCPU, stats.Cpu.Usage.PerCpu[i])
+			tags := map[string]string{"instance": fmt.Sprintf("%v", i)}
+			addTagsToPoint(point, tags)
 
-		points = append(points, point)
-	}
+			points = append(points, point)
+		}
 
-	// Load Average
-	points = append(points, makePoint(serLoadAverage, stats.Cpu.LoadAverage))
+		// Load Average
+		points = append(points, makePoint(serLoadAverage, stats.Cpu.LoadAverage))
+	}
 
 	// Network Stats
-	points = append(points, makePoint(serRxBytes, stats.Network.RxBytes))
-	points = append(points, makePoint(serRxErrors, stats.Network.RxErrors))
-	points = append(points, makePoint(serTxBytes, stats.Network.TxBytes))
-	points = append(points, makePoint(serTxErrors, stats.Network.TxErrors))
+	if stats.Network != nil {
+		points = append(points, makePoint(serRxBytes, stats.Network.RxBytes))
+		points = append(points, makePoint(serRxErrors, stats.Network.RxErrors))
+		points = append(points, makePoint(serTxBytes, stats.Network.TxBytes))
+		points = append(points, makePoint(serTxErrors, stats.Network.TxErrors))
+	}
 
 	// Referenced Memory
 	points = append(points, makePoint(serReferencedMemory, stats.ReferencedMemory))
@@ -246,6 +265,9 @@ func (s *influxdbStorage) memoryStatsToPoints(
 	cInfo *info.ContainerInfo,
 	stats *info.ContainerStats,
 ) (points []*influxdb.Point) {
+	if stats.Memory == nil {
+		return points
+	}
 	// Memory Usage
 	points = append(points, makePoint(serMemoryUsage, stats.Memory.Usage))
 	// Maximum memory usage recorded
@@ -414,42 +436,73 @@ func (s *influxdbStorage) AddStats(cInfo *info.ContainerInfo, stats *info.Contai
 		}
 	}()
 	if len(pointsToFlush) > 0 {
-		points := make([]influxdb.Point, len(pointsToFlush))
-		for i, p := range pointsToFlush {
-			points[i] = *p
+		if err := s.flush(pointsToFlush, stats.Timestamp); err != nil {
+			return err
 		}
+	}
+	return nil
+}
 
-		batchTags := map[string]string{tagMachineName: s.machineName}
-		bp := influxdb.BatchPoints{
-			Points:          points,
-			Database:        s.database,
-			RetentionPolicy: s.retentionPolicy,
-			Tags:            batchTags,
-			Time:            stats.Timestamp,
+func (s *influxdbStorage) flush(pointsToFlush []*influxdb.Point, timestamp time.Time) error {
+	if s.writeAPI != nil {
+		// Token-authenticated mode: write through the InfluxDB 2.x API,
+		// which is also implemented by InfluxDB 3 Core.
+		points := make([]*write.Point, len(pointsToFlush))
+		for i, p := range pointsToFlush {
+			points[i] = influxdb2.NewPoint(p.Measurement, p.Tags, p.Fields, p.Time)
 		}
-		response, err := s.client.Write(bp)
-		if err != nil || checkResponseForErrors(response) != nil {
+		if err := s.writeAPI.WritePoint(context.Background(), points...); err != nil {
 			return fmt.Errorf("failed to write stats to influxDb - %s", err)
 		}
+		return nil
+	}
+
+	points := make([]influxdb.Point, len(pointsToFlush))
+	for i, p := range pointsToFlush {
+		points[i] = *p
+	}
+
+	batchTags := map[string]string{tagMachineName: s.machineName}
+	bp := influxdb.BatchPoints{
+		Points:          points,
+		Database:        s.database,
+		RetentionPolicy: s.retentionPolicy,
+		Tags:            batchTags,
+		Time:            timestamp,
+	}
+	response, err := s.client.Write(bp)
+	if err != nil || checkResponseForErrors(response) != nil {
+		return fmt.Errorf("failed to write stats to influxDb - %s", err)
 	}
 	return nil
 }
 
 func (s *influxdbStorage) Close() error {
+	if s.v2Client != nil {
+		s.v2Client.Close()
+	}
 	s.client = nil
+	s.v2Client = nil
+	s.writeAPI = nil
 	return nil
 }
 
 // machineName: A unique identifier to identify the host that current cAdvisor
 // instance is running on.
 // influxdbHost: The host which runs influxdb (host:port)
+// authToken: When non-empty, stats are written with token authentication
+// through the InfluxDB 2.x API (also implemented by InfluxDB 3 Core) and
+// username/password are ignored. The bucket falls back to database when empty.
 func newStorage(
 	machineName,
 	tablename,
 	database,
 	retentionPolicy,
 	username,
 	password,
+	authToken,
+	org,
+	bucket,
 	influxdbHost string,
 	isSecure bool,
 	bufferDuration time.Duration,
@@ -462,26 +515,34 @@ func newStorage(
 		url.Scheme = "https"
 	}
 
-	config := &influxdb.Config{
-		URL:       *url,
-		Username:  username,
-		Password:  password,
-		UserAgent: fmt.Sprintf("%v/%v", "cAdvisor", version.Info["version"]),
-	}
-	client, err := influxdb.NewClient(*config)
-	if err != nil {
-		return nil, err
-	}
-
 	ret := &influxdbStorage{
-		client:          client,
 		machineName:     machineName,
 		database:        database,
 		retentionPolicy: retentionPolicy,
 		bufferDuration:  bufferDuration,
 		lastWrite:       time.Now(),
 		points:          make([]*influxdb.Point, 0),
 	}
+	if authToken != "" {
+		if bucket == "" {
+			bucket = database
+		}
+		ret.bucket = bucket
+		ret.v2Client = influxdb2.NewClient(url.String(), authToken)
+		ret.writeAPI = ret.v2Client.WriteAPIBlocking(org, bucket)
+	} else {
+		config := &influxdb.Config{
+			URL:       *url,
+			Username:  username,
+			Password:  password,
+			UserAgent: fmt.Sprintf("%v/%v", "cAdvisor", version.Info["version"]),
+		}
+		client, err := influxdb.NewClient(*config)
+		if err != nil {
+			return nil, err
+		}
+		ret.client = client
+	}
 	ret.readyToFlush = ret.defaultReadyToFlush
 	return ret, nil
 }