Skip to content

Commit

Permalink
Add pprof and create admin endpoint (jaegertracing#1315)
Browse files Browse the repository at this point in the history
  • Loading branch information
Konrad Galuszka committed Feb 26, 2019
1 parent adec968 commit e791ce6
Show file tree
Hide file tree
Showing 10 changed files with 235 additions and 99 deletions.
7 changes: 4 additions & 3 deletions cmd/all-in-one/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,8 @@ import (
"github.com/jaegertracing/jaeger/cmd/flags"
queryApp "github.com/jaegertracing/jaeger/cmd/query/app"
"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
"github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
"github.com/jaegertracing/jaeger/pkg/version"
Expand Down Expand Up @@ -102,9 +102,10 @@ func main() {
if err != nil {
return err
}
hc, err := sFlags.NewHealthCheck(logger)
hc := sFlags.NewHealthCheck(logger)
err = sFlags.NewAdminEndpoint(logger, hc)
if err != nil {
logger.Fatal("Could not start the health check server.", zap.Error(err))
logger.Fatal("Could not start the admin endpoint server.", zap.Error(err))
}

mBldr := new(pMetrics.Builder).InitFromViper(v)
Expand Down
7 changes: 4 additions & 3 deletions cmd/collector/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,8 @@ import (
"github.com/jaegertracing/jaeger/cmd/collector/app/zipkin"
"github.com/jaegertracing/jaeger/cmd/env"
"github.com/jaegertracing/jaeger/cmd/flags"
"github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
"github.com/jaegertracing/jaeger/pkg/version"
Expand Down Expand Up @@ -85,9 +85,10 @@ func main() {
if err != nil {
return err
}
hc, err := sFlags.NewHealthCheck(logger)
hc := sFlags.NewHealthCheck(logger)
err = sFlags.NewAdminEndpoint(logger, hc)
if err != nil {
logger.Fatal("Could not start the health check server.", zap.Error(err))
logger.Fatal("Could not start the admin endpoint server.", zap.Error(err))
}

builderOpts := new(builder.CollectorOptions).InitFromViper(v)
Expand Down
28 changes: 17 additions & 11 deletions cmd/flags/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,8 @@ import (
"go.uber.org/zap"
"go.uber.org/zap/zapcore"

hc "github.com/jaegertracing/jaeger/pkg/healthcheck"
ae "github.com/jaegertracing/jaeger/pkg/adminendpoint"
hc "github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/plugin/storage"
)

Expand Down Expand Up @@ -57,15 +58,15 @@ func TryLoadConfigFile(v *viper.Viper) error {
type SharedFlags struct {
// Logging holds logging configuration
Logging logging
// HealthCheck holds health check configuration
HealthCheck healthCheck
// AdminEndpoint holds admin endpoint configuration
AdminEndpoint adminendpoint
}

type logging struct {
Level string
}

type healthCheck struct {
type adminendpoint struct {
Port int
}

Expand All @@ -89,7 +90,7 @@ func AddLoggingFlag(flagSet *flag.FlagSet) {
// InitFromViper initializes SharedFlags with properties from viper
func (flags *SharedFlags) InitFromViper(v *viper.Viper) *SharedFlags {
flags.Logging.Level = v.GetString(logLevel)
flags.HealthCheck.Port = v.GetInt(healthCheckHTTPPort)
flags.AdminEndpoint.Port = v.GetInt(healthCheckHTTPPort)
return flags
}

Expand All @@ -104,11 +105,16 @@ func (flags *SharedFlags) NewLogger(conf zap.Config, options ...zap.Option) (*za
return conf.Build(options...)
}

// NewHealthCheck returns health check based on configuration in SharedFlags
func (flags *SharedFlags) NewHealthCheck(logger *zap.Logger) (*hc.HealthCheck, error) {
if flags.HealthCheck.Port == 0 {
return nil, errors.New("port not specified")
// NewHealthCheck returns health check
func (flags *SharedFlags) NewHealthCheck(logger *zap.Logger) *hc.HealthCheck {
return hc.New(hc.Unavailable, hc.Logger(logger))
}

// NewAdminEndpoint returns admin endpoint based on configuration in SharedFlags
func (flags *SharedFlags) NewAdminEndpoint(logger *zap.Logger, healthcheck *hc.HealthCheck) error {
if flags.AdminEndpoint.Port == 0 {
return errors.New("port not specified")
}
return hc.New(hc.Unavailable, hc.Logger(logger)).
Serve(flags.HealthCheck.Port)
return ae.New(ae.Logger(logger), ae.HealthCheck(healthcheck)).
Serve(flags.AdminEndpoint.Port)
}
7 changes: 4 additions & 3 deletions cmd/ingester/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,8 @@ import (
"github.com/jaegertracing/jaeger/cmd/flags"
"github.com/jaegertracing/jaeger/cmd/ingester/app"
"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
"github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
"github.com/jaegertracing/jaeger/pkg/version"
Expand Down Expand Up @@ -67,9 +67,10 @@ func main() {
if err != nil {
return err
}
hc, err := sFlags.NewHealthCheck(logger)
hc := sFlags.NewHealthCheck(logger)
err = sFlags.NewAdminEndpoint(logger, hc)
if err != nil {
logger.Fatal("Could not start the health check server.", zap.Error(err))
logger.Fatal("Could not start the admin endpoint server.", zap.Error(err))
}

mBldr := new(pMetrics.Builder).InitFromViper(v)
Expand Down
7 changes: 4 additions & 3 deletions cmd/query/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,8 @@ import (
"github.com/jaegertracing/jaeger/cmd/flags"
"github.com/jaegertracing/jaeger/cmd/query/app"
"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
"github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/config"
"github.com/jaegertracing/jaeger/pkg/healthcheck"
pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
"github.com/jaegertracing/jaeger/pkg/recoveryhandler"
"github.com/jaegertracing/jaeger/pkg/version"
Expand Down Expand Up @@ -72,9 +72,10 @@ func main() {
if err != nil {
return err
}
hc, err := sFlags.NewHealthCheck(logger)
hc := sFlags.NewHealthCheck(logger)
err = sFlags.NewAdminEndpoint(logger, hc)
if err != nil {
logger.Fatal("Could not start the health check server.", zap.Error(err))
logger.Fatal("Could not start the admin endpoint server.", zap.Error(err))
}

queryOpts := new(app.QueryOptions).InitFromViper(v)
Expand Down
120 changes: 120 additions & 0 deletions pkg/adminendpoint/adminendpoint.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
// Copyright (c) 2017 Uber Technologies, 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 adminendpoint

import (
"context"
"net"
"net/http"
"net/http/pprof"
"strconv"

"go.uber.org/zap"

hc "github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/version"
)

// HealthCheck provides an HTTP endpoint that returns the health status of the service
type AdminEndpoint struct {
logger *zap.Logger
server *http.Server
healthcheck *hc.HealthCheck
}

// Option is a functional option for passing parameters to New()
type Option func(endpoint *AdminEndpoint)

// Logger creates an option to set the logger. If not specified, Nop logger is used.
func Logger(logger *zap.Logger) Option {
return func(ae *AdminEndpoint) {
ae.logger = logger
}
}

// HealthCheck creates an option to set the healthcheck. Required
func HealthCheck(healthcheck *hc.HealthCheck) Option {
return func(ae *AdminEndpoint) {
ae.healthcheck = healthcheck
}
}

// New creates a AdminEndpoint
func New(options ...Option) *AdminEndpoint {
ae := &AdminEndpoint{}
for _, option := range options {
option(ae)
}
if ae.logger == nil {
ae.logger = zap.NewNop()
}
return ae
}

// Serve starts HTTP server on the specified port.
func (ae *AdminEndpoint) Serve(port int) error {
portStr := ":" + strconv.Itoa(port)
l, err := net.Listen("tcp", portStr)
if err != nil {
ae.logger.Error("Admin endpoint server failed to listen", zap.Error(err))
return err
}
ae.serveWithListener(l)
ae.logger.Info("Admin endpoint server started", zap.Int("http-port", port), zap.Stringer("status", ae.healthcheck.Get()))
return nil
}

// serveWithListener starts server using given listener
func (ae *AdminEndpoint) serveWithListener(l net.Listener) {
ae.server = &http.Server{Handler: ae.httpHandler()}
go func() {
if err := ae.server.Serve(l); err != nil {
ae.logger.Error("failed to serve", zap.Error(err))
ae.healthcheck.Set(hc.Broken)
}
}()
}

// Close stops the HTTP server
func (ae *AdminEndpoint) Close() error {
return ae.server.Shutdown(context.Background())
}

// httpHandler creates a new HTTP handler.
func (ae *AdminEndpoint) httpHandler() http.Handler {
mux := http.NewServeMux()
ae.registerHealthCheckHandler(mux)
ae.registerProfilingHandler(mux)
version.RegisterHandler(mux, ae.logger)
return mux
}

// registerHealthCheckHandler registers
func (ae *AdminEndpoint) registerHealthCheckHandler(mux *http.ServeMux) {
mux.HandleFunc("/", ae.healthcheck.GetHandlerFunc())
}

// registerProfilingHandler adds pprof endpoints
func (ae *AdminEndpoint) registerProfilingHandler(mux *http.ServeMux) {
mux.HandleFunc("/debug/pprof/", pprof.Index)
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
mux.Handle("/debug/pprof/goroutine", pprof.Handler("goroutine"))
mux.Handle("/debug/pprof/heap", pprof.Handler("heap"))
mux.Handle("/debug/pprof/threadcreate", pprof.Handler("threadcreate"))
mux.Handle("/debug/pprof/block", pprof.Handler("block"))
}
47 changes: 47 additions & 0 deletions pkg/adminendpoint/adminendpoint_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
// Copyright (c) 2017 Uber Technologies, 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 adminendpoint_test

import (
"net"
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

"github.com/jaegertracing/jaeger/pkg/adminendpoint"
"github.com/jaegertracing/jaeger/pkg/adminendpoint/healthcheck"
"github.com/jaegertracing/jaeger/pkg/testutils"
)

func TestPortBusy(t *testing.T) {
l, err := net.Listen("tcp", ":0")
require.NoError(t, err)
defer l.Close()
port := l.Addr().(*net.TCPAddr).Port

logger, logBuf := testutils.NewLogger()
err = adminendpoint.New(adminendpoint.Logger(logger)).Serve(port)
assert.Error(t, err)
assert.Equal(t, "Health Check server failed to listen", logBuf.JSONLine(0)["msg"])
}

func TestServeHandler(t *testing.T) {
hc := healthcheck.New(healthcheck.Ready)
ae := adminendpoint.New(adminendpoint.HealthCheck(hc))
err := ae.Serve(0)
require.NoError(t, err)
defer ae.Close()
}
Loading

0 comments on commit e791ce6

Please sign in to comment.