Add an option to store payload to files. (#124)

This change introduces an option to store the CreateTimeSeriesRequest
protobuf messages in wire format to files under a directory specified,
instead of sending protobuf messages to Stackdriver Monitoring API.

Author: StevenYCChou

cmd/stackdriver-prometheus-sidecar/main.go

@@ -180,24 +180,25 @@ func main() {
 	}
 
 	cfg := struct {
-		configFilename     string
-		projectIdResource  string
-		kubernetesLabels   kubernetesConfig
-		genericLabels      genericConfig
-		stackdriverAddress *url.URL
-		metricsPrefix      string
-		useGkeResource     bool
-		walDirectory       string
-		prometheusURL      *url.URL
-		listenAddress      string
-		filters            []string
-		filtersets         []string
-		aggregations       retrieval.CounterAggregatorConfig
-		metricRenames      map[string]string
-		staticMetadata     []scrape.MetricMetadata
-		useRestrictedIps   bool
-		manualResolver     *manual.Resolver
-		monitoringBackends []string
+		configFilename        string
+		projectIdResource     string
+		kubernetesLabels      kubernetesConfig
+		genericLabels         genericConfig
+		stackdriverAddress    *url.URL
+		metricsPrefix         string
+		useGkeResource        bool
+		storeInFilesDirectory string
+		walDirectory          string
+		prometheusURL         *url.URL
+		listenAddress         string
+		filters               []string
+		filtersets            []string
+		aggregations          retrieval.CounterAggregatorConfig
+		metricRenames         map[string]string
+		staticMetadata        []scrape.MetricMetadata
+		useRestrictedIps      bool
+		manualResolver        *manual.Resolver
+		monitoringBackends    []string
 
 		logLevel promlog.AllowedLevel
 	}{}
@@ -239,6 +240,9 @@ func main() {
 		"Whether to use the legacy gke_container MonitoredResource type instead of k8s_container").
 		Default("false").BoolVar(&cfg.useGkeResource)
 
+	a.Flag("stackdriver.store-in-files-directory", "If specified, store the CreateTimeSeriesRequest protobuf messages to files under this directory, instead of sending protobuf messages to Stackdriver Monitoring API.").
+		StringVar(&cfg.storeInFilesDirectory)
+
 	a.Flag("prometheus.wal-directory", "Directory from where to read the Prometheus TSDB WAL.").
 		Default("data/wal").StringVar(&cfg.walDirectory)
 
@@ -393,16 +397,34 @@ func main() {
 	// works well.
 	config.DefaultQueueConfig.Capacity = 3 * stackdriver.MaxTimeseriesesPerRequest
 
-	queueManager, err := stackdriver.NewQueueManager(
-		log.With(logger, "component", "queue_manager"),
-		config.DefaultQueueConfig,
-		&clientFactory{
+	var scf stackdriver.StorageClientFactory
+
+	if len(cfg.storeInFilesDirectory) > 0 {
+		err := os.MkdirAll(cfg.storeInFilesDirectory, 0700)
+		if err != nil {
+			level.Error(logger).Log(
+				"msg", "Failure creating directory.",
+				"err", err)
+			os.Exit(1)
+		}
+		scf = &fileClientFactory{
+			dir:    cfg.storeInFilesDirectory,
+			logger: log.With(logger, "component", "storage"),
+		}
+	} else {
+		scf = &stackdriverClientFactory{
 			logger:            log.With(logger, "component", "storage"),
 			projectIdResource: cfg.projectIdResource,
 			url:               cfg.stackdriverAddress,
 			timeout:           10 * time.Second,
 			manualResolver:    cfg.manualResolver,
-		},
+		}
+	}
+
+	queueManager, err := stackdriver.NewQueueManager(
+		log.With(logger, "component", "queue_manager"),
+		config.DefaultQueueConfig,
+		scf,
 		tailer,
 	)
 	if err != nil {
@@ -561,26 +583,50 @@ func main() {
 	level.Info(logger).Log("msg", "See you next time!")
 }
 
-type clientFactory struct {
+type stackdriverClientFactory struct {
 	logger            log.Logger
 	projectIdResource string
 	url               *url.URL
 	timeout           time.Duration
 	manualResolver    *manual.Resolver
 }
 
-func (f *clientFactory) New() stackdriver.StorageClient {
+func (s *stackdriverClientFactory) New() stackdriver.StorageClient {
 	return stackdriver.NewClient(&stackdriver.ClientConfig{
-		Logger:    f.logger,
-		ProjectId: f.projectIdResource,
-		URL:       f.url,
-		Timeout:   f.timeout,
-		Resolver:  f.manualResolver,
+		Logger:    s.logger,
+		ProjectId: s.projectIdResource,
+		URL:       s.url,
+		Timeout:   s.timeout,
+		Resolver:  s.manualResolver,
 	})
 }
 
-func (f *clientFactory) Name() string {
-	return f.url.String()
+func (s *stackdriverClientFactory) Name() string {
+	return s.url.String()
+}
+
+// fileClientFactory generates StorageClient which writes to a newly
+// created file under dir. It requires dir an existing valid directory.
+type fileClientFactory struct {
+	dir    string
+	logger log.Logger
+}
+
+// New creates an instance of stackdriver.StorageClient. Each instance
+// writes to a different file under dir. The returned instance is not
+// thread-safe.
+func (fcf *fileClientFactory) New() stackdriver.StorageClient {
+	f, err := ioutil.TempFile(fcf.dir, "*.txt")
+	if err != nil {
+		level.Error(fcf.logger).Log(
+			"msg", "failure creating files.",
+			"err", err)
+	}
+	return stackdriver.NewCreateTimeSeriesRequestWriterCloser(f, fcf.logger)
+}
+
+func (f *fileClientFactory) Name() string {
+	return "fileClientFactory"
 }
 
 func waitForPrometheus(ctx context.Context, logger log.Logger, promURL *url.URL) {

stackdriver/writer.go

@@ -0,0 +1,67 @@
+/*
+Copyright 2019 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package stackdriver
+
+import (
+	"io"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/golang/protobuf/proto"
+	monitoring "google.golang.org/genproto/googleapis/monitoring/v3"
+)
+
+// CreateTimeSeriesRequestWriterCloser allows writing protobuf message
+// monitoring.CreateTimeSeriesRequest as wire format into the writerCloser.
+type CreateTimeSeriesRequestWriterCloser struct {
+	logger      log.Logger
+	writeCloser io.WriteCloser
+}
+
+func NewCreateTimeSeriesRequestWriterCloser(writeCloser io.WriteCloser, logger log.Logger) *CreateTimeSeriesRequestWriterCloser {
+	if logger == nil {
+		logger = log.NewNopLogger()
+	}
+	return &CreateTimeSeriesRequestWriterCloser{
+		writeCloser: writeCloser,
+		logger:      logger,
+	}
+}
+
+// Store writes protobuf message monitoring.CreateTimeSeriesRequest as wire
+// format into the writeCloser.
+func (c *CreateTimeSeriesRequestWriterCloser) Store(req *monitoring.CreateTimeSeriesRequest) error {
+	data, err := proto.Marshal(req)
+	if err != nil {
+		level.Warn(c.logger).Log(
+			"msg", "failure marshaling CreateTimeSeriesRequest.",
+			"err", err)
+		return err
+	}
+	_, err = c.writeCloser.Write(data)
+	if err != nil {
+		level.Warn(c.logger).Log(
+			"msg", "failure writing data to file.",
+			"err", err)
+		return err
+	}
+	return nil
+}
+
+func (c *CreateTimeSeriesRequestWriterCloser) Close() error {
+	return c.writeCloser.Close()
+}

stackdriver/writer_test.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2019 Google Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package stackdriver
+
+import (
+	"bytes"
+	"reflect"
+	"testing"
+
+	"github.com/go-kit/kit/log"
+	"github.com/golang/protobuf/proto"
+	monitoring "google.golang.org/genproto/googleapis/monitoring/v3"
+)
+
+type myWriterCloser struct {
+	Buffer bytes.Buffer
+}
+
+func (m *myWriterCloser) Write(p []byte) (int, error) {
+	return m.Buffer.Write(p)
+}
+
+func (m *myWriterCloser) Close() error {
+	m.Buffer.Reset()
+	return nil
+}
+
+func TestRequest(t *testing.T) {
+	var m myWriterCloser
+	c := NewCreateTimeSeriesRequestWriterCloser(&m, log.NewNopLogger())
+	defer c.Close()
+	req := &monitoring.CreateTimeSeriesRequest{
+		TimeSeries: []*monitoring.TimeSeries{
+			&monitoring.TimeSeries{},
+		},
+	}
+	if err := c.Store(req); err != nil {
+		t.Fatal(err)
+	}
+
+	storedReq := &monitoring.CreateTimeSeriesRequest{}
+	err := proto.Unmarshal(m.Buffer.Bytes(), storedReq)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(req, storedReq) {
+		t.Fatalf("Expect requests as %v, but stored as: %v", req, storedReq)
+	}
+}