cmd/ocagent: retrofitted with new protocol

With this change, cmd/ocagent now adheres to the new protocol and currently runs the OpenCensus interceptor while exporting received spans to the various and previously already created YAML configured exporters such as: * Stackdriver Tracing * DataDog * Zipkin which are statically imported. Now any program that has an ocagent-exporter can send traces to this service. The service's OpenCensus Interceptor Port can be configured by ```shell ocagent -oci-port=<value> ``` which by default is `55678` After this change we can add both more trace exporters and more trace interceptors. Since the metrics proto and design hasn't yet been finalized, let's remove any vestiges of metrics in the various exporters for now but can trivially add them back whenever we are ready. Essentially this can be our v0.0.1 of the new trace agent/service.
census-instrumentation · Oct 3, 2018 · 0c052cc · 0c052cc
1 parent 086a878
commit 0c052cc
Show file tree

Hide file tree

Showing 5 changed files with 145 additions and 152 deletions.
diff --git a/cmd/ocagent/exporter/datadog.go → cmd/ocagent/exporterparser/datadog.go b/cmd/ocagent/exporter/datadog.go → cmd/ocagent/exporterparser/datadog.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+package exporterparser
 
 import (
 	"log"
@@ -24,23 +24,22 @@ import (
 )
 
 type dataDogConfig struct {
-	Datadog struct {
+	Datadog *struct {
 		// Namespace specifies the namespaces to which metric keys are appended.
 		Namespace string `yaml:"namespace,omitempty"`
 
 		// TraceAddr specifies the host[:port] address of the Datadog Trace Agent.
 		// It defaults to localhost:8126.
-		TraceAddr string `yaml:"traceAddr,omitempty"`
+		TraceAddr string `yaml:"trace_addr,omitempty"`
 
 		// MetricsAddr specifies the host[:port] address for DogStatsD. It defaults
 		// to localhost:8125.
-		MetricsAddr string `yaml:"metricsAddr,omitempty"`
+		MetricsAddr string `yaml:"metrics_addr,omitempty"`
 
 		// Tags specifies a set of global tags to attach to each metric.
 		Tags []string `yaml:"tags,omitempty"`
 
-		EnableMetrics bool `yaml:"enableMetrics,omitempty"`
-		EnableTraces  bool `yaml:"enableTraces,omitempty"`
+		EnableTracing bool `yaml:"enable_tracing,omitempty"`
 	} `yaml:"datadog,omitempty"`
 }
 
@@ -51,21 +50,22 @@ func (d *datadogExporter) MakeExporters(config []byte) (se view.Exporter, te tra
 	if err := yaml.Unmarshal(config, &c); err != nil {
 		log.Fatalf("Cannot unmarshal data: %v", err)
 	}
-	if c := c.Datadog; c.EnableMetrics || c.EnableTraces {
-		// TODO(jbd): Create new exporter for each service name.
-		de := datadog.NewExporter(datadog.Options{
-			Namespace: c.Namespace,
-			TraceAddr: c.TraceAddr,
-			StatsAddr: c.MetricsAddr,
-			Tags:      c.Tags,
-		})
-		if c.EnableMetrics {
-			se = de
-		}
-		if c.EnableTraces {
-			te = de
-		}
-		closer = de.Stop
+
+	dc := c.Datadog
+	if dc == nil {
+		return nil, nil, nil
+	}
+	if !dc.EnableTracing {
+		return nil, nil, nil
 	}
-	return se, te, closer
+
+	// TODO(jbd): Create new exporter for each service name.
+	de := datadog.NewExporter(datadog.Options{
+		Namespace: dc.Namespace,
+		TraceAddr: dc.TraceAddr,
+		StatsAddr: dc.MetricsAddr,
+		Tags:      dc.Tags,
+	})
+	closer = de.Stop
+	return nil, de, closer
 }
diff --git a/cmd/ocagent/exporter/exporter.go → cmd/ocagent/exporterparser/exparser.go b/cmd/ocagent/exporter/exporter.go → cmd/ocagent/exporterparser/exparser.go
@@ -12,19 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+// This package provides support for parsing and creating the
+// respective exporters given a YAML configuration payload.
+// For now it currently only provides statically imported OpenCensus
+// exporters like:
+//  * Stackdriver Tracing and Monitoring
+//  * DataDog
+//  * Zipkin
+
+package exporterparser
 
 import (
+	"sync"
+
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/trace"
 )
 
 var (
-	exporters []Exporter
-
-	statsExporters []view.Exporter
-	traceExporters []trace.Exporter
-	closers        []func()
+	exportersMu sync.Mutex
+	exporters   []Exporter
 )
 
 func init() {
@@ -37,16 +44,17 @@ type Exporter interface {
 	MakeExporters(config []byte) (se view.Exporter, te trace.Exporter, closer func())
 }
 
-// RegisterExporter allows users to export additional exporters
-// before configuration is parsed. RegisterExporter should be called
-// before Parse.
+// RegisterExporter allows users to dyanmically add additional exporters
+// before the configuration is parsed. RegisterExporter should be called
+// before ExportersFromYAMLConfig.
 func RegisterExporter(e Exporter) {
+	exportersMu.Lock()
 	exporters = append(exporters, e)
+	exportersMu.Unlock()
 }
 
-// Parse parses the config yaml payload and configures
-// the exporters accordingly.
-func Parse(config []byte) {
+// ExportersFromYAMLConfig parses the config yaml payload and returns the respective exporters
+func ExportersFromYAMLConfig(config []byte) (traceExporters []trace.Exporter, statsExporters []view.Exporter, closeFns []func()) {
 	for _, e := range exporters {
 		se, te, closer := e.MakeExporters(config)
 		if se != nil {
@@ -56,26 +64,8 @@ func Parse(config []byte) {
 			traceExporters = append(traceExporters, te)
 		}
 		if closer != nil {
-			closers = append(closers, closer)
+			closeFns = append(closeFns, closer)
 		}
 	}
-}
-
-// ExportView exports the view data to all registered view exporters.
-func ExportView(vs *view.Data) {
-	// TODO(jbd): Implement.
-}
-
-// ExportSpan exports the span data to all registered trace exporters.
-func ExportSpan(sd *trace.SpanData) {
-	for _, exporter := range traceExporters {
-		exporter.ExportSpan(sd)
-	}
-}
-
-// CloseAll closes all of the exporters.
-func CloseAll() {
-	for _, fn := range closers {
-		fn()
-	}
+	return
 }
diff --git a/cmd/ocagent/exporter/stackdriver.go → cmd/ocagent/exporterparser/stackdriver.go b/cmd/ocagent/exporter/stackdriver.go → cmd/ocagent/exporterparser/stackdriver.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+package exporterparser
 
 import (
 	"log"
@@ -24,10 +24,9 @@ import (
 )
 
 type stackdriverConfig struct {
-	Stackdriver struct {
+	Stackdriver *struct {
 		ProjectID     string `yaml:"project,omitempty"`
-		EnableMetrics bool   `yaml:"enableMetrics,omitempty"`
-		EnableTraces  bool   `yaml:"enableTraces,omitempty"`
+		EnableTracing bool   `yaml:"enable_tracing,omitempty"`
 	} `yaml:"stackdriver,omitempty"`
 }
 
@@ -38,24 +37,25 @@ func (s *stackdriverExporter) MakeExporters(config []byte) (se view.Exporter, te
 	if err := yaml.Unmarshal(config, &c); err != nil {
 		log.Fatalf("Cannot unmarshal data: %v", err)
 	}
-	if s := c.Stackdriver; s.EnableMetrics || s.EnableTraces {
-		// TODO(jbd): Add monitored resources.
-		if s.ProjectID == "" {
-			log.Fatal("Stackdriver config requires a project ID")
-		}
-		exporter, err := stackdriver.NewExporter(stackdriver.Options{
-			ProjectID: s.ProjectID,
-		})
-		if err != nil {
-			log.Fatalf("Cannot configure Stackdriver exporter: %v", err)
-		}
-		if s.EnableMetrics {
-			se = exporter
-		}
-		if s.EnableTraces {
-			te = exporter
-		}
-		closer = exporter.Flush
+	sc := c.Stackdriver
+	if sc == nil {
+		return nil, nil, nil
 	}
-	return se, te, closer
+	if !sc.EnableTracing {
+		return nil, nil, nil
+	}
+
+	// TODO(jbd): Add monitored resources.
+	if sc.ProjectID == "" {
+		log.Fatal("Stackdriver config requires a project ID")
+	}
+	sde, err := stackdriver.NewExporter(stackdriver.Options{
+		ProjectID: sc.ProjectID,
+	})
+	if err != nil {
+		log.Fatalf("Cannot configure Stackdriver exporter: %v", err)
+	}
+
+	closer = sde.Flush
+	return nil, sde, closer
 }
diff --git a/cmd/ocagent/exporter/zipkin.go → cmd/ocagent/exporterparser/zipkin.go b/cmd/ocagent/exporter/zipkin.go → cmd/ocagent/exporterparser/zipkin.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+package exporterparser
 
 import (
 	"log"
@@ -26,8 +26,10 @@ import (
 )
 
 type zipkinConfig struct {
-	Zipkin struct {
-		Endpoint string `yaml:"endpoint,omitempty"`
+	Zipkin *struct {
+		ServiceName      string `yaml:"service_name,omitempty"`
+		Endpoint         string `yaml:"endpoint,omitempty"`
+		LocalEndpointURI string `yaml:"local_endpoint,omitempty"`
 	} `yaml:"zipkin,omitempty"`
 }
 
@@ -38,18 +40,33 @@ func (z *zipkinExporter) MakeExporters(config []byte) (se view.Exporter, te trac
 	if err := yaml.Unmarshal(config, &c); err != nil {
 		log.Fatalf("Cannot unmarshal data: %v", err)
 	}
-	if endpoint := c.Zipkin.Endpoint; endpoint != "" {
-		// TODO(jbd): Propagate service name, hostport and more metadata from each node.
-		localEndpoint, err := openzipkin.NewEndpoint("", "")
-		if err != nil {
-			log.Fatalf("Cannot configure Zipkin exporter: %v", err)
-		}
-		reporter := http.NewReporter(endpoint)
-		te = zipkin.NewExporter(reporter, localEndpoint)
-		closer = func() {
-			if err := reporter.Close(); err != nil {
-				log.Printf("Cannot close the Zipkin reporter: %v\n", err)
-			}
+	if c.Zipkin == nil {
+		return nil, nil, nil
+	}
+	zc := c.Zipkin
+	endpoint := "http://localhost:9411/api/v2/spans"
+	if zc.Endpoint != "" {
+		endpoint = zc.Endpoint
+	}
+	serviceName := ""
+	if zc.ServiceName != "" {
+		serviceName = zc.ServiceName
+	}
+	localEndpointURI := "192.168.1.5:5454"
+	if zc.LocalEndpointURI != "" {
+		localEndpointURI = zc.LocalEndpointURI
+	}
+	// TODO(jbd): Propagate hostport and more metadata from each node.
+	localEndpoint, err := openzipkin.NewEndpoint(serviceName, localEndpointURI)
+	if err != nil {
+		log.Fatalf("Cannot configure Zipkin exporter: %v", err)
+	}
+
+	reporter := http.NewReporter(endpoint)
+	te = zipkin.NewExporter(reporter, localEndpoint)
+	closer = func() {
+		if err := reporter.Close(); err != nil {
+			log.Printf("Cannot close the Zipkin reporter: %v\n", err)
 		}
 	}
 	return se, te, closer