Add zpages to collector (#541)

* Add zpages to collector

Move zpages to its own type and share the usage of that type between agent and collector.

* Paying coverage tax

* Added very basic end-to-end coverage for collector
* A new test for zpagesserver: port already in use

* Improve collector test synchronization

* Make TestApplication_Start resilient to port conflicts

There are other tests in the code base that can suffer with port conflicts but I didn't wat to go overboard and chase them right now.

Author: Paulo Janotti

cmd/ocagent/main.go

@@ -20,16 +20,13 @@ import (
 	"context"
 	"fmt"
 	"log"
-	"net"
-	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"go.opencensus.io/stats/view"
-	"go.opencensus.io/zpages"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -38,6 +35,7 @@ import (
 	"github.com/census-instrumentation/opencensus-service/internal/config/viperutils"
 	"github.com/census-instrumentation/opencensus-service/internal/pprofserver"
 	"github.com/census-instrumentation/opencensus-service/internal/version"
+	"github.com/census-instrumentation/opencensus-service/internal/zpagesserver"
 	"github.com/census-instrumentation/opencensus-service/observability"
 	"github.com/census-instrumentation/opencensus-service/processor/multiconsumer"
 	"github.com/census-instrumentation/opencensus-service/receiver/jaegerreceiver"
@@ -132,7 +130,11 @@ func runOCAgent() {
 	// If zPages are enabled, run them
 	zPagesPort, zPagesEnabled := agentConfig.ZPagesPort()
 	if zPagesEnabled {
-		zCloseFn := runZPages(zPagesPort)
+		zCloseFn, err := zpagesserver.Run(asyncErrorChan, zPagesPort)
+		if err != nil {
+			log.Fatal(err)
+		}
+		log.Printf("Running zPages on port %d", zPagesPort)
 		closeFns = append(closeFns, zCloseFn)
 	}
 
@@ -202,28 +204,6 @@ func runOCAgent() {
 	}
 }
 
-func runZPages(port int) func() error {
-	// And enable zPages too
-	zPagesMux := http.NewServeMux()
-	zpages.Handle(zPagesMux, "/debug")
-
-	addr := fmt.Sprintf(":%d", port)
-	ln, err := net.Listen("tcp", addr)
-	if err != nil {
-		log.Fatalf("Failed to bind to run zPages on %q: %v", addr, err)
-	}
-
-	srv := http.Server{Handler: zPagesMux}
-	go func() {
-		log.Printf("Running zPages at %q", addr)
-		if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {
-			log.Fatalf("Failed to serve zPages: %v", err)
-		}
-	}()
-
-	return srv.Close
-}
-
 func runOCReceiver(logger *zap.Logger, acfg *config.Config, tc consumer.TraceConsumer, mc consumer.MetricsConsumer, asyncErrorChan chan<- error) (doneFn func() error, err error) {
 	tlsCredsOption, hasTLSCreds, err := acfg.OpenCensusReceiverTLSCredentialsServerOption()
 	if err != nil {

cmd/occollector/app/collector/collector.go

@@ -32,6 +32,7 @@ import (
 	"github.com/census-instrumentation/opencensus-service/consumer"
 	"github.com/census-instrumentation/opencensus-service/internal/config/viperutils"
 	"github.com/census-instrumentation/opencensus-service/internal/pprofserver"
+	"github.com/census-instrumentation/opencensus-service/internal/zpagesserver"
 	"github.com/census-instrumentation/opencensus-service/receiver"
 )
 
@@ -47,11 +48,16 @@ type Application struct {
 	healthCheck *healthcheck.HealthCheck
 	processor   consumer.TraceConsumer
 	receivers   []receiver.TraceReceiver
+	// stopTestChan is used to terminate the application in end to end tests.
+	stopTestChan chan struct{}
+	// readyChan is used in tests to indicate that the application is ready.
+	readyChan chan struct{}
 }
 
 func newApp() *Application {
 	return &Application{
-		v: viper.New(),
+		v:         viper.New(),
+		readyChan: make(chan struct{}),
 	}
 }
 
@@ -89,6 +95,20 @@ func (app *Application) execute() {
 	var closeFns []func()
 	app.processor, closeFns = startProcessor(app.v, app.logger)
 
+	zpagesPort := app.v.GetInt(zpagesserver.ZPagesHTTPPort)
+	if zpagesPort > 0 {
+		closeZPages, err := zpagesserver.Run(asyncErrorChannel, zpagesPort)
+		if err != nil {
+			app.logger.Error("Failed to run zPages", zap.Error(err))
+			os.Exit(1)
+		}
+		app.logger.Info("Running zPages", zap.Int("port", zpagesPort))
+		closeFn := func() {
+			closeZPages()
+		}
+		closeFns = append(closeFns, closeFn)
+	}
+
 	app.receivers = createReceivers(app.v, app.logger, app.processor, asyncErrorChannel)
 
 	err = initTelemetry(asyncErrorChannel, app.v, app.logger)
@@ -103,11 +123,18 @@ func (app *Application) execute() {
 	// mark service as ready to receive traffic.
 	app.healthCheck.Ready()
 
+	// set the channel to stop testing.
+	app.stopTestChan = make(chan struct{})
+	// notify tests that it is ready.
+	close(app.readyChan)
+
 	select {
 	case err = <-asyncErrorChannel:
 		app.logger.Error("Asynchronous error received, terminating process", zap.Error(err))
 	case s := <-signalsChannel:
 		app.logger.Info("Received signal from OS", zap.String("signal", s.String()))
+	case <-app.stopTestChan:
+		app.logger.Info("Received stop test request")
 	}
 
 	app.healthCheck.Set(healthcheck.Unavailable)
@@ -144,6 +171,7 @@ func (app *Application) Start() error {
 		healthCheckFlags,
 		loggerFlags,
 		pprofserver.AddFlags,
+		zpagesserver.AddFlags,
 	)
 
 	return rootCmd.Execute()

cmd/occollector/app/collector/collector_test.go

@@ -0,0 +1,90 @@
+// Copyright 2019, 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 collector handles the command-line, configuration, and runs the OC collector.
+package collector
+
+import (
+	"net"
+	"net/http"
+	"testing"
+
+	"github.com/census-instrumentation/opencensus-service/internal/zpagesserver"
+)
+
+func TestApplication_Start(t *testing.T) {
+	portArg := []string{
+		healthCheckHTTPPort, // Keep it as first since its address is used later.
+		zpagesserver.ZPagesHTTPPort,
+		"metrics-port",
+		"receivers.opencensus.port",
+	}
+	addresses := getMultipleAvailableLocalAddresses(t, uint(len(portArg)))
+	for i, addr := range addresses {
+		_, port, err := net.SplitHostPort(addr)
+		if err != nil {
+			t.Fatalf("failed to split host and port from %q: %v", addr, err)
+		}
+		App.v.Set(portArg[i], port)
+	}
+
+	// Without exporters the collector will start and just shutdown, no error is expected.
+	App.v.Set("logging-exporter", true)
+
+	appDone := make(chan struct{})
+	go func() {
+		defer close(appDone)
+		if err := App.Start(); err != nil {
+			t.Fatalf("App.Start() got %v, want nil", err)
+		}
+	}()
+
+	<-App.readyChan
+	if !isAppAvailable(t, "http://"+addresses[0]) {
+		t.Fatalf("App didn't reach ready state")
+	}
+	close(App.stopTestChan)
+	<-appDone
+}
+
+// isAppAvailable checks if the healthcheck server at the given endpoint is
+// returning `available`.
+func isAppAvailable(t *testing.T, healthCheckEndPoint string) bool {
+	client := &http.Client{}
+	resp, err := client.Get(healthCheckEndPoint)
+	if err != nil {
+		t.Fatalf("failed to get a response from health probe: %v", err)
+	}
+	defer resp.Body.Close()
+	return resp.StatusCode == http.StatusNoContent
+}
+
+func getMultipleAvailableLocalAddresses(t *testing.T, numAddresses uint) []string {
+	addresses := make([]string, numAddresses, numAddresses)
+	for i := uint(0); i < numAddresses; i++ {
+		addresses[i] = getAvailableLocalAddress(t)
+	}
+	return addresses
+}
+
+func getAvailableLocalAddress(t *testing.T) string {
+	ln, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("failed to get a free local port: %v", err)
+	}
+	// There is a possible race if something else takes this same port before
+	// the test uses it, however, that is unlikely in practice.
+	defer ln.Close()
+	return ln.Addr().String()
+}

cmd/occollector/app/collector/empty_test.go

@@ -22,6 +22,7 @@ services:
       - ./oc-collector-config.yaml:/etc/oc-collector-config.yaml
     ports:
       - "55678"
+      - "55680:55679"
       - "1777:1777"
       - "8888:8888" # Prometheus metrics
     depends_on:
@@ -38,6 +39,7 @@ services:
       - "1888:1888"
       - "14268"
       - "55678"
+      - "55679:55679"
     depends_on:
       - oc-collector
 

demos/trace/docker-compose.yaml

@@ -0,0 +1,58 @@
+// Copyright 2019, 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 zpagesserver
+
+import (
+	"flag"
+	"fmt"
+	"net"
+	"net/http"
+
+	"go.opencensus.io/zpages"
+)
+
+const (
+	// ZPagesHTTPPort is the name of the flag used to specify the zpages port.
+	ZPagesHTTPPort = "zpages-http-port"
+)
+
+// AddFlags adds to the flag set a flag to configure the zpages server.
+func AddFlags(flags *flag.FlagSet) {
+	flags.Uint(
+		ZPagesHTTPPort,
+		55679,
+		"Port on which to run the zpages http server, use 0 to disable zpages.")
+}
+
+// Run run a zPages HTTP endpoint on the given port.
+func Run(asyncErrorChannel chan<- error, port int) (closeFn func() error, err error) {
+	zPagesMux := http.NewServeMux()
+	zpages.Handle(zPagesMux, "/debug")
+
+	addr := fmt.Sprintf(":%d", port)
+	ln, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to bind to run zPages on %q: %v", addr, err)
+	}
+
+	srv := http.Server{Handler: zPagesMux}
+	go func() {
+		if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed {
+			asyncErrorChannel <- fmt.Errorf("failed to server zPages: %v", err)
+		}
+	}()
+
+	return srv.Close, nil
+}