Add processor queue

CONTRIBUTING.md

@@ -53,3 +53,13 @@ $ git checkout -b feature
 $ git commit
 $ git push fork feature
 ```
+
+## General Notes
+
+This project uses go 1.11 and Travis for CI.
+
+Travis CI uses the Makefile with the default target, it is recommended to
+run it before submitting your PR. It runs `gofmt -s` (simplify) and `golint`.
+
+The dependencies are managed with `go mod` if you work with the sources under your
+`$GOPATH` you need to set the environment variable `GO111MODULE=on`.

README.md

@@ -292,20 +292,38 @@ The collector is in its initial development stages. It can be run directly
 from sources, binary, or a Docker image.
 
 1. Run from sources:
-```
+```shell
 $ go run github.com/census-instrumentation/opencensus-service/cmd/occollector
 ```
 2. Run from binary (from the root of your repo):
-```
+```shell
 $ make collector
 $ ./bin/occollector_$($GOOS)
 ```
 3. Build a Docker scratch image and use the appropria Docker command for your scenario:
-```
+```shell
 $ make docker-collector
 $ docker run --rm -it -p 55678:55678 occollector
 ```
 
+4. It can be configured via command-line or config file:
+```shell
+OpenCensus Collector
+
+Usage:
+  occollector [flags]
+
+Flags:
+      --add-queued-processor   Flag to wrap one processor with the queued processor (flag will be remove soon, dev helper)
+      --config string          Path to the config file
+  -h, --help                   help for occollector
+      --log-level string       Output level of logs (TRACE, DEBUG, INFO, WARN, ERROR, FATAL) (default "INFO")
+      --noop-processor         Flag to add the no-op processor (combine with log level DEBUG to log incoming spans)
+      --receive-jaeger         Flag to run the Jaeger receiver (i.e.: Jaeger Collector), default settings: {ThriftTChannelPort:14267 ThriftHTTPPort:14268}
+      --receive-oc-trace       Flag to run the OpenCensus trace receiver, default settings: {Port:55678}
+      --receive-zipkin         Flag to run the Zipkin receiver, default settings: {Port:9411}
+```
+
 [travis-image]: https://travis-ci.org/census-instrumentation/opencensus-service.svg?branch=master
 [travis-url]: https://travis-ci.org/census-instrumentation/opencensus-service
 [godoc-image]: https://godoc.org/github.com/census-instrumentation/opencensus-service?status.svg

cmd/occollector/app/builder/builder.go

@@ -0,0 +1,138 @@
+// Copyright 2018, OpenCensus Authors
+//
+// 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 builder handles the options to build the OpenCensus collector
+// pipeline.
+package builder
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/spf13/viper"
+)
+
+const (
+	receiversRoot   = "receivers"
+	jaegerEntry     = "jaeger"
+	opencensusEntry = "opencensus"
+	zipkinEntry     = "zipkin"
+)
+
+// JaegerReceiverCfg holds configuration for Jaeger receivers.
+type JaegerReceiverCfg struct {
+	// ThriftTChannelPort is the port that the relay receives on for jaeger thrift tchannel requests
+	ThriftTChannelPort int `mapstructure:"thrift-tchannel-port"`
+	// ThriftHTTPPort is the port that the relay receives on for jaeger thrift http requests
+	ThriftHTTPPort int `mapstructure:"thrift-http-port"`
+}
+
+// JaegerReceiverEnabled checks if the Jaeger receiver is enabled, via a command-line flag, environment
+// variable, or configuration file.
+func JaegerReceiverEnabled(v *viper.Viper, cmdFlag string) bool {
+	return featureEnabled(v, cmdFlag, receiversRoot, jaegerEntry)
+}
+
+// NewDefaultJaegerReceiverCfg returns an instance of JaegerReceiverCfg with default values
+func NewDefaultJaegerReceiverCfg() *JaegerReceiverCfg {
+	opts := &JaegerReceiverCfg{
+		ThriftTChannelPort: 14267,
+		ThriftHTTPPort:     14268,
+	}
+	return opts
+}
+
+// InitFromViper returns a JaegerReceiverCfg according to the configuration.
+func (cfg *JaegerReceiverCfg) InitFromViper(v *viper.Viper) (*JaegerReceiverCfg, error) {
+	return cfg, initFromViper(cfg, v, receiversRoot, jaegerEntry)
+}
+
+// OpenCensusReceiverCfg holds configuration for OpenCensus receiver.
+type OpenCensusReceiverCfg struct {
+	// Port is the port that the receiver will use
+	Port int `mapstructure:"port"`
+}
+
+// OpenCensusReceiverEnabled checks if the OpenCensus receiver is enabled, via a command-line flag, environment
+// variable, or configuration file.
+func OpenCensusReceiverEnabled(v *viper.Viper, cmdFlag string) bool {
+	return featureEnabled(v, cmdFlag, receiversRoot, opencensusEntry)
+}
+
+// NewDefaultOpenCensusReceiverCfg returns an instance of OpenCensusReceiverCfg with default values
+func NewDefaultOpenCensusReceiverCfg() *OpenCensusReceiverCfg {
+	opts := &OpenCensusReceiverCfg{
+		Port: 55678,
+	}
+	return opts
+}
+
+// InitFromViper returns a OpenCensusReceiverCfg according to the configuration.
+func (cfg *OpenCensusReceiverCfg) InitFromViper(v *viper.Viper) (*OpenCensusReceiverCfg, error) {
+	return cfg, initFromViper(cfg, v, receiversRoot, opencensusEntry)
+}
+
+// ZipkinReceiverCfg holds configuration for Zipkin receiver.
+type ZipkinReceiverCfg struct {
+	// Port is the port that the receiver will use
+	Port int `mapstructure:"port"`
+}
+
+// ZipkinReceiverEnabled checks if the Zipkin receiver is enabled, via a command-line flag, environment
+// variable, or configuration file.
+func ZipkinReceiverEnabled(v *viper.Viper, cmdFlag string) bool {
+	return featureEnabled(v, cmdFlag, receiversRoot, zipkinEntry)
+}
+
+// NewDefaultZipkinReceiverCfg returns an instance of ZipkinReceiverCfg with default values
+func NewDefaultZipkinReceiverCfg() *ZipkinReceiverCfg {
+	opts := &ZipkinReceiverCfg{
+		Port: 9411,
+	}
+	return opts
+}
+
+// InitFromViper returns a ZipkinReceiverCfg according to the configuration.
+func (cfg *ZipkinReceiverCfg) InitFromViper(v *viper.Viper) (*ZipkinReceiverCfg, error) {
+	return cfg, initFromViper(cfg, v, receiversRoot, zipkinEntry)
+}
+
+// Helper functions
+
+func initFromViper(cfg interface{}, v *viper.Viper, labels ...string) error {
+	v = getViperSub(v, labels...)
+	if v == nil {
+		return nil
+	}
+	if err := v.Unmarshal(cfg); err != nil {
+		return fmt.Errorf("Failed to read configuration for %s %v", strings.Join(labels, ": "), err)
+	}
+
+	return nil
+}
+
+func getViperSub(v *viper.Viper, labels ...string) *viper.Viper {
+	for _, label := range labels {
+		v = v.Sub(label)
+		if v == nil {
+			return nil
+		}
+	}
+
+	return v
+}
+
+func featureEnabled(v *viper.Viper, cmdFlag string, labels ...string) bool {
+	return v.GetBool(cmdFlag) || (getViperSub(v, labels...) != nil)
+}

cmd/occollector/app/builder/builder_test.go

@@ -0,0 +1,80 @@
+// Copyright 2018, OpenCensus Authors
+//
+// 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 builder
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/spf13/viper"
+)
+
+func TestReceiversEnabledByPresenceWithDefaultSettings(t *testing.T) {
+	v, err := loadViperFromFile("./testdata/receivers_enabled.yaml")
+	if err != nil {
+		t.Fatalf("Failed to load viper from test file: %v", err)
+	}
+
+	jaegerEnabled, opencensusEnabled, zipkinEnabled := JaegerReceiverEnabled(v, "j"), OpenCensusReceiverEnabled(v, "oc"), ZipkinReceiverEnabled(v, "z")
+	if !jaegerEnabled || !opencensusEnabled || !zipkinEnabled {
+		t.Fatalf("Some of the expected receivers were not enabled j:%v oc:%v z:%v", jaegerEnabled, opencensusEnabled, zipkinEnabled)
+	}
+
+	wj := NewDefaultJaegerReceiverCfg()
+	gj, err := wj.InitFromViper(v)
+	if err != nil {
+		t.Errorf("Failed to InitFromViper for Jaeger receiver: %v", err)
+	} else if !reflect.DeepEqual(wj, gj) {
+		t.Errorf("Incorrect config for Jaeger receiver, want %v got %v", wj, gj)
+	}
+
+	woc := NewDefaultOpenCensusReceiverCfg()
+	goc, err := woc.InitFromViper(v)
+	if err != nil {
+		t.Errorf("Failed to InitFromViper for OpenCensus receiver: %v", err)
+	} else if !reflect.DeepEqual(woc, goc) {
+		t.Errorf("Incorrect config for OpenCensus receiver, want %v got %v", woc, goc)
+	}
+
+	wz := NewDefaultZipkinReceiverCfg()
+	gz, err := wz.InitFromViper(v)
+	if err != nil {
+		t.Errorf("Failed to InitFromViper for Zipkin receiver: %v", err)
+	} else if !reflect.DeepEqual(wz, gz) {
+		t.Errorf("Incorrect config for Zipkin receiver, want %v got %v", wz, gz)
+	}
+}
+
+func TestReceiversDisabledByPresenceWithDefaultSettings(t *testing.T) {
+	v, err := loadViperFromFile("./testdata/receivers_disabled.yaml")
+	if err != nil {
+		t.Fatalf("Failed to load viper from test file: %v", err)
+	}
+
+	jaegerEnabled, opencensusEnabled, zipkinEnabled := JaegerReceiverEnabled(v, "j"), OpenCensusReceiverEnabled(v, "oc"), ZipkinReceiverEnabled(v, "z")
+	if jaegerEnabled || opencensusEnabled || zipkinEnabled {
+		t.Fatalf("Not all receivers were disabled j:%v oc:%v z:%v", jaegerEnabled, opencensusEnabled, zipkinEnabled)
+	}
+}
+
+func loadViperFromFile(file string) (*viper.Viper, error) {
+	v := viper.New()
+	v.SetConfigFile(file)
+	err := v.ReadInConfig()
+	if err != nil {
+		return nil, err
+	}
+	return v, nil
+}

cmd/occollector/app/builder/testdata/receivers_disabled.yaml

@@ -0,0 +1,5 @@
+# No receivers enabled, all commented out
+receivers:
+  # jaeger: {} 
+  # opencensus: {}
+  # zipkin: {}

cmd/occollector/app/builder/testdata/receivers_enabled.yaml

@@ -0,0 +1,5 @@
+# Enable receivers with default configuration
+receivers:
+  jaeger: {} 
+  opencensus: {}
+  zipkin: {}