mirror of
https://github.com/pomerium/pomerium.git
synced 2025-06-13 16:24:16 +02:00
envoy: Enable zipkin tracing (#737)
- Update envoy bootstrap config to protobufs - Reorganize tracing config to avoid cyclic import - Push down zipkin config to Envoy - Update tracing options to provide sample rate
This commit is contained in:
parent
38c1b5ec65
commit
3e17befff7
13 changed files with 434 additions and 120 deletions
|
@ -34,6 +34,9 @@ const DisableHeaderKey = "disable"
|
||||||
// gRPC server, or is used for healthchecks (authorize only service)
|
// gRPC server, or is used for healthchecks (authorize only service)
|
||||||
const DefaultAlternativeAddr = ":5443"
|
const DefaultAlternativeAddr = ":5443"
|
||||||
|
|
||||||
|
// EnvoyAdminURL indicates where the envoy control plane is listening
|
||||||
|
var EnvoyAdminURL = &url.URL{Host: "localhost:9901", Scheme: "http"}
|
||||||
|
|
||||||
// Options are the global environmental flags used to set up pomerium's services.
|
// Options are the global environmental flags used to set up pomerium's services.
|
||||||
// Use NewXXXOptions() methods for a safely initialized data structure.
|
// Use NewXXXOptions() methods for a safely initialized data structure.
|
||||||
type Options struct {
|
type Options struct {
|
||||||
|
@ -172,7 +175,7 @@ type Options struct {
|
||||||
|
|
||||||
// Tracing shared settings
|
// Tracing shared settings
|
||||||
TracingProvider string `mapstructure:"tracing_provider" yaml:"tracing_provider,omitempty"`
|
TracingProvider string `mapstructure:"tracing_provider" yaml:"tracing_provider,omitempty"`
|
||||||
TracingDebug bool `mapstructure:"tracing_debug" yaml:"tracing_debug,omitempty"`
|
TracingSampleRate float64 `mapstructure:"tracing_sample_rate" yaml:"tracing_sample_rate,omitempty"`
|
||||||
|
|
||||||
// Jaeger
|
// Jaeger
|
||||||
//
|
//
|
||||||
|
@ -272,6 +275,7 @@ var defaultOptions = Options{
|
||||||
CacheStore: "autocache",
|
CacheStore: "autocache",
|
||||||
AuthenticateCallbackPath: "/oauth2/callback",
|
AuthenticateCallbackPath: "/oauth2/callback",
|
||||||
AutoCertFolder: dataDir(),
|
AutoCertFolder: dataDir(),
|
||||||
|
TracingSampleRate: 0.0001,
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewDefaultOptions returns a copy the default options. It's the caller's
|
// NewDefaultOptions returns a copy the default options. It's the caller's
|
||||||
|
|
|
@ -206,7 +206,7 @@ func Test_Checksum(t *testing.T) {
|
||||||
func TestOptionsFromViper(t *testing.T) {
|
func TestOptionsFromViper(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
opts := []cmp.Option{
|
opts := []cmp.Option{
|
||||||
cmpopts.IgnoreFields(Options{}, "CacheStore", "CookieSecret", "GRPCInsecure", "GRPCAddr", "CacheURLString", "CacheURL", "AuthorizeURL", "AuthorizeURLString", "DefaultUpstreamTimeout", "CookieExpire", "Services", "Addr", "RefreshCooldown", "LogLevel", "KeyFile", "CertFile", "SharedKey", "ReadTimeout", "IdleTimeout", "GRPCClientTimeout", "GRPCClientDNSRoundRobin"),
|
cmpopts.IgnoreFields(Options{}, "CacheStore", "CookieSecret", "GRPCInsecure", "GRPCAddr", "CacheURLString", "CacheURL", "AuthorizeURL", "AuthorizeURLString", "DefaultUpstreamTimeout", "CookieExpire", "Services", "Addr", "RefreshCooldown", "LogLevel", "KeyFile", "CertFile", "SharedKey", "ReadTimeout", "IdleTimeout", "GRPCClientTimeout", "GRPCClientDNSRoundRobin", "TracingSampleRate"),
|
||||||
cmpopts.IgnoreFields(Policy{}, "Source", "Destination"),
|
cmpopts.IgnoreFields(Policy{}, "Source", "Destination"),
|
||||||
cmpOptIgnoreUnexported,
|
cmpOptIgnoreUnexported,
|
||||||
}
|
}
|
||||||
|
|
80
config/trace.go
Normal file
80
config/trace.go
Normal file
|
@ -0,0 +1,80 @@
|
||||||
|
package config
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"net/url"
|
||||||
|
|
||||||
|
"github.com/pomerium/pomerium/internal/urlutil"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
// JaegerTracingProviderName is the name of the tracing provider Jaeger.
|
||||||
|
JaegerTracingProviderName = "jaeger"
|
||||||
|
// ZipkinTracingProviderName is the name of the tracing provider Zipkin.
|
||||||
|
ZipkinTracingProviderName = "zipkin"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TracingOptions contains the configurations settings for a http server.
|
||||||
|
type TracingOptions struct {
|
||||||
|
// Shared
|
||||||
|
Provider string
|
||||||
|
Service string
|
||||||
|
Debug bool
|
||||||
|
|
||||||
|
// Jaeger
|
||||||
|
|
||||||
|
// CollectorEndpoint is the full url to the Jaeger HTTP Thrift collector.
|
||||||
|
// For example, http://localhost:14268/api/traces
|
||||||
|
JaegerCollectorEndpoint *url.URL
|
||||||
|
// AgentEndpoint instructs exporter to send spans to jaeger-agent at this address.
|
||||||
|
// For example, localhost:6831.
|
||||||
|
JaegerAgentEndpoint string
|
||||||
|
|
||||||
|
// Zipkin
|
||||||
|
|
||||||
|
// ZipkinEndpoint configures the zipkin collector URI
|
||||||
|
// Example: http://zipkin:9411/api/v2/spans
|
||||||
|
ZipkinEndpoint *url.URL
|
||||||
|
|
||||||
|
// SampleRate is percentage of requests which are sampled
|
||||||
|
SampleRate float64
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewTracingOptions builds a new TracingOptions from core Options
|
||||||
|
func NewTracingOptions(o *Options) (*TracingOptions, error) {
|
||||||
|
tracingOpts := TracingOptions{
|
||||||
|
Provider: o.TracingProvider,
|
||||||
|
Service: o.Services,
|
||||||
|
JaegerAgentEndpoint: o.TracingJaegerAgentEndpoint,
|
||||||
|
SampleRate: o.TracingSampleRate,
|
||||||
|
}
|
||||||
|
|
||||||
|
switch o.TracingProvider {
|
||||||
|
case JaegerTracingProviderName:
|
||||||
|
if o.TracingJaegerCollectorEndpoint != "" {
|
||||||
|
jaegerCollectorEndpoint, err := urlutil.ParseAndValidateURL(o.TracingJaegerCollectorEndpoint)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("config: invalid jaeger endpoint url: %w", err)
|
||||||
|
}
|
||||||
|
tracingOpts.JaegerCollectorEndpoint = jaegerCollectorEndpoint
|
||||||
|
tracingOpts.JaegerAgentEndpoint = o.TracingJaegerAgentEndpoint
|
||||||
|
}
|
||||||
|
case ZipkinTracingProviderName:
|
||||||
|
zipkinEndpoint, err := urlutil.ParseAndValidateURL(o.ZipkinEndpoint)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("config: invalid zipkin endpoint url: %w", err)
|
||||||
|
}
|
||||||
|
tracingOpts.ZipkinEndpoint = zipkinEndpoint
|
||||||
|
case "":
|
||||||
|
default:
|
||||||
|
return nil, fmt.Errorf("config: provider %s unknown", o.TracingProvider)
|
||||||
|
}
|
||||||
|
|
||||||
|
return &tracingOpts, nil
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
// Enabled indicates whether tracing is enabled on a given TracingOptions
|
||||||
|
func (t *TracingOptions) Enabled() bool {
|
||||||
|
return t.Provider != ""
|
||||||
|
}
|
79
config/trace_test.go
Normal file
79
config/trace_test.go
Normal file
|
@ -0,0 +1,79 @@
|
||||||
|
package config
|
||||||
|
|
||||||
|
import (
|
||||||
|
"net/url"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/google/go-cmp/cmp"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
func Test_NewTracingOptions(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
opts *Options
|
||||||
|
want *TracingOptions
|
||||||
|
wantErr bool
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
"jaeger_good",
|
||||||
|
&Options{TracingProvider: "jaeger", TracingJaegerAgentEndpoint: "foo", TracingJaegerCollectorEndpoint: "http://foo"},
|
||||||
|
&TracingOptions{Provider: "jaeger", JaegerAgentEndpoint: "foo", JaegerCollectorEndpoint: &url.URL{Scheme: "http", Host: "foo"}},
|
||||||
|
false,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"jaeger_bad",
|
||||||
|
&Options{TracingProvider: "jaeger", TracingJaegerAgentEndpoint: "foo", TracingJaegerCollectorEndpoint: "badurl"},
|
||||||
|
nil,
|
||||||
|
true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"zipkin_good",
|
||||||
|
&Options{TracingProvider: "zipkin", ZipkinEndpoint: "https://foo/api/v1/spans"},
|
||||||
|
&TracingOptions{Provider: "zipkin", ZipkinEndpoint: &url.URL{Scheme: "https", Host: "foo", Path: "/api/v1/spans"}},
|
||||||
|
false,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"zipkin_bad",
|
||||||
|
&Options{TracingProvider: "zipkin", ZipkinEndpoint: "notaurl"},
|
||||||
|
nil,
|
||||||
|
true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"noprovider",
|
||||||
|
&Options{},
|
||||||
|
&TracingOptions{},
|
||||||
|
false,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"fakeprovider",
|
||||||
|
&Options{TracingProvider: "fake"},
|
||||||
|
nil,
|
||||||
|
true,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
got, err := NewTracingOptions(tt.opts)
|
||||||
|
assert.NotEqual(t, err == nil, tt.wantErr, "unexpected error value")
|
||||||
|
assert.Empty(t, cmp.Diff(tt.want, got))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func Test_TracingEnabled(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
opts *TracingOptions
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"enabled", &TracingOptions{Provider: "zipkin"}, true},
|
||||||
|
{"not enabled", &TracingOptions{}, false},
|
||||||
|
}
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
assert.Equal(t, tt.want, tt.opts.Enabled(), "unexpected tracing state")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
|
@ -450,9 +450,9 @@ Each unit work is called a Span in a trace. Spans include metadata about the wor
|
||||||
#### Shared Tracing Settings
|
#### Shared Tracing Settings
|
||||||
|
|
||||||
| Config Key | Description | Required |
|
| Config Key | Description | Required |
|
||||||
| :--------------- | :---------------------------------------------------------------- | -------- |
|
| :------------------ | :------------------------------------------------------------------------------------ | -------- |
|
||||||
| tracing_provider | The name of the tracing provider. (e.g. jaeger, zipkin) | ✅ |
|
| tracing_provider | The name of the tracing provider. (e.g. jaeger, zipkin) | ✅ |
|
||||||
| tracing_debug | Will disable [sampling](https://opencensus.io/tracing/sampling/). | ❌ |
|
| tracing_sample_rate | Percentage of requests to sample in decimal notation. Default is `0.0001`, or `.01%` | ❌ |
|
||||||
|
|
||||||
#### Jaeger (partial)
|
#### Jaeger (partial)
|
||||||
|
|
||||||
|
|
|
@ -11,7 +11,8 @@ description: >-
|
||||||
|
|
||||||
### Tracing
|
### Tracing
|
||||||
|
|
||||||
Jaeger tracing support is no longer end-to-end in the proxy service. We recommend updating to the Zipkin provider for proper tracing support. Jaeger will continue to work but will not have coverage in the data plane.
|
- Jaeger tracing support is no longer end-to-end in the proxy service. We recommend updating to the Zipkin provider for proper tracing support. Jaeger will continue to work but will not have coverage in the data plane.
|
||||||
|
- Option `tracing_debug` is no longer supported. Use `tracing_sampling_rate` instead. [Details](https://www.pomerium.io/configuration/#shared-tracing-settings).
|
||||||
|
|
||||||
# Since 0.7.0
|
# Since 0.7.0
|
||||||
|
|
||||||
|
|
|
@ -62,9 +62,9 @@ func Run(ctx context.Context, configFile string) error {
|
||||||
_, httpPort, _ := net.SplitHostPort(controlPlane.HTTPListener.Addr().String())
|
_, httpPort, _ := net.SplitHostPort(controlPlane.HTTPListener.Addr().String())
|
||||||
|
|
||||||
// create envoy server
|
// create envoy server
|
||||||
envoyServer, err := envoy.NewServer(grpcPort, httpPort)
|
envoyServer, err := envoy.NewServer(opt, grpcPort, httpPort)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("error creating envoy server")
|
return fmt.Errorf("error creating envoy server: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// add services
|
// add services
|
||||||
|
@ -165,7 +165,7 @@ func setupCache(opt *config.Options, controlPlane *controlplane.Server) error {
|
||||||
|
|
||||||
func setupMetrics(ctx context.Context, opt *config.Options) error {
|
func setupMetrics(ctx context.Context, opt *config.Options) error {
|
||||||
if opt.MetricsAddr != "" {
|
if opt.MetricsAddr != "" {
|
||||||
handler, err := metrics.PrometheusHandler()
|
handler, err := metrics.PrometheusHandler(config.EnvoyAdminURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
@ -203,16 +203,12 @@ func setupProxy(opt *config.Options, controlPlane *controlplane.Server) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
func setupTracing(ctx context.Context, opt *config.Options) error {
|
func setupTracing(ctx context.Context, opt *config.Options) error {
|
||||||
if opt.TracingProvider != "" {
|
traceOpts, err := config.NewTracingOptions(opt)
|
||||||
tracingOpts := &trace.TracingOptions{
|
if err != nil {
|
||||||
Provider: opt.TracingProvider,
|
return fmt.Errorf("error setting up tracing: %w", err)
|
||||||
Service: opt.Services,
|
|
||||||
Debug: opt.TracingDebug,
|
|
||||||
JaegerAgentEndpoint: opt.TracingJaegerAgentEndpoint,
|
|
||||||
JaegerCollectorEndpoint: opt.TracingJaegerCollectorEndpoint,
|
|
||||||
ZipkinEndpoint: opt.ZipkinEndpoint,
|
|
||||||
}
|
}
|
||||||
exporter, err := trace.RegisterTracing(tracingOpts)
|
if traceOpts.Enabled() {
|
||||||
|
exporter, err := trace.RegisterTracing(traceOpts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
|
@ -217,6 +217,9 @@ func (srv *Server) buildMainHTTPConnectionManagerFilter(options *config.Options,
|
||||||
MaxStreamDuration: maxStreamDuration,
|
MaxStreamDuration: maxStreamDuration,
|
||||||
},
|
},
|
||||||
RequestTimeout: ptypes.DurationProto(options.ReadTimeout),
|
RequestTimeout: ptypes.DurationProto(options.ReadTimeout),
|
||||||
|
Tracing: &envoy_http_connection_manager.HttpConnectionManager_Tracing{
|
||||||
|
RandomSampling: &envoy_type_v3.Percent{Value: options.TracingSampleRate * 100},
|
||||||
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
return &envoy_config_listener_v3.Filter{
|
return &envoy_config_listener_v3.Filter{
|
||||||
|
|
|
@ -2,6 +2,8 @@
|
||||||
package envoy
|
package envoy
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
|
|
||||||
"bufio"
|
"bufio"
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
@ -12,19 +14,26 @@ import (
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"regexp"
|
"regexp"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
|
||||||
|
|
||||||
|
envoy_config_bootstrap_v3 "github.com/envoyproxy/go-control-plane/envoy/config/bootstrap/v3"
|
||||||
|
envoy_config_cluster_v3 "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
|
||||||
|
envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
|
||||||
|
envoy_config_endpoint_v3 "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
|
||||||
|
envoy_config_trace_v3 "github.com/envoyproxy/go-control-plane/envoy/config/trace/v3"
|
||||||
|
"github.com/golang/protobuf/proto"
|
||||||
|
"github.com/golang/protobuf/ptypes"
|
||||||
"github.com/natefinch/atomic"
|
"github.com/natefinch/atomic"
|
||||||
"github.com/rs/zerolog"
|
"github.com/rs/zerolog"
|
||||||
|
"google.golang.org/protobuf/encoding/protojson"
|
||||||
|
"google.golang.org/protobuf/types/known/durationpb"
|
||||||
|
|
||||||
|
"github.com/pomerium/pomerium/config"
|
||||||
"github.com/pomerium/pomerium/internal/log"
|
"github.com/pomerium/pomerium/internal/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
workingDirectoryName = ".pomerium-envoy"
|
workingDirectoryName = ".pomerium-envoy"
|
||||||
configFileName = "envoy-config.yaml"
|
configFileName = "envoy-config.yaml"
|
||||||
// EnvoyAdminURL indicates where the envoy control plane is listening
|
|
||||||
EnvoyAdminURL = "http://localhost:9901"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// A Server is a pomerium proxy implemented via envoy.
|
// A Server is a pomerium proxy implemented via envoy.
|
||||||
|
@ -33,10 +42,11 @@ type Server struct {
|
||||||
cmd *exec.Cmd
|
cmd *exec.Cmd
|
||||||
|
|
||||||
grpcPort, httpPort string
|
grpcPort, httpPort string
|
||||||
|
opts *config.Options
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewServer creates a new server with traffic routed by envoy.
|
// NewServer creates a new server with traffic routed by envoy.
|
||||||
func NewServer(grpcPort, httpPort string) (*Server, error) {
|
func NewServer(opts *config.Options, grpcPort, httpPort string) (*Server, error) {
|
||||||
wd := filepath.Join(os.TempDir(), workingDirectoryName)
|
wd := filepath.Join(os.TempDir(), workingDirectoryName)
|
||||||
err := os.MkdirAll(wd, 0755)
|
err := os.MkdirAll(wd, 0755)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -47,6 +57,7 @@ func NewServer(grpcPort, httpPort string) (*Server, error) {
|
||||||
wd: wd,
|
wd: wd,
|
||||||
grpcPort: grpcPort,
|
grpcPort: grpcPort,
|
||||||
httpPort: httpPort,
|
httpPort: httpPort,
|
||||||
|
opts: opts,
|
||||||
}
|
}
|
||||||
|
|
||||||
err = srv.writeConfig()
|
err = srv.writeConfig()
|
||||||
|
@ -96,40 +107,206 @@ func (srv *Server) Run(ctx context.Context) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (srv *Server) writeConfig() error {
|
func (srv *Server) writeConfig() error {
|
||||||
return atomic.WriteFile(filepath.Join(srv.wd, configFileName), strings.NewReader(`
|
confBytes, err := srv.buildBootstrapConfig()
|
||||||
node:
|
if err != nil {
|
||||||
id: pomerium-envoy
|
return err
|
||||||
cluster: pomerium-envoy
|
}
|
||||||
|
|
||||||
admin:
|
cfgPath := filepath.Join(srv.wd, configFileName)
|
||||||
access_log_path: /tmp/admin_access.log
|
log.WithLevel(zerolog.DebugLevel).Str("service", "envoy").Str("location", cfgPath).Msg("wrote config file to location")
|
||||||
address:
|
|
||||||
socket_address: { address: 127.0.0.1, port_value: 9901 }
|
|
||||||
|
|
||||||
dynamic_resources:
|
return atomic.WriteFile(cfgPath, bytes.NewReader(confBytes))
|
||||||
cds_config:
|
}
|
||||||
ads: {}
|
|
||||||
resource_api_version: V3
|
func (srv *Server) buildBootstrapConfig() ([]byte, error) {
|
||||||
lds_config:
|
|
||||||
ads: {}
|
nodeCfg := &envoy_config_core_v3.Node{
|
||||||
resource_api_version: V3
|
Id: "proxy",
|
||||||
ads_config:
|
Cluster: "proxy",
|
||||||
api_type: GRPC
|
}
|
||||||
transport_api_version: V3
|
|
||||||
grpc_services:
|
adminCfg := &envoy_config_bootstrap_v3.Admin{
|
||||||
- envoy_grpc:
|
AccessLogPath: "/tmp/admin_access.log",
|
||||||
cluster_name: pomerium-control-plane-grpc
|
Address: &envoy_config_core_v3.Address{
|
||||||
static_resources:
|
Address: &envoy_config_core_v3.Address_SocketAddress{
|
||||||
clusters:
|
SocketAddress: &envoy_config_core_v3.SocketAddress{
|
||||||
- name: pomerium-control-plane-grpc
|
Address: "127.0.0.1",
|
||||||
connect_timeout: { seconds: 5 }
|
PortSpecifier: &envoy_config_core_v3.SocketAddress_PortValue{
|
||||||
type: STATIC
|
PortValue: 9901,
|
||||||
hosts:
|
},
|
||||||
- socket_address:
|
},
|
||||||
address: 127.0.0.1
|
},
|
||||||
port_value: `+srv.grpcPort+`
|
},
|
||||||
http2_protocol_options: {}
|
}
|
||||||
`))
|
|
||||||
|
dynamicCfg := &envoy_config_bootstrap_v3.Bootstrap_DynamicResources{
|
||||||
|
AdsConfig: &envoy_config_core_v3.ApiConfigSource{
|
||||||
|
ApiType: envoy_config_core_v3.ApiConfigSource_ApiType(envoy_config_core_v3.ApiConfigSource_ApiType_value["GRPC"]),
|
||||||
|
TransportApiVersion: envoy_config_core_v3.ApiVersion_V3,
|
||||||
|
GrpcServices: []*envoy_config_core_v3.GrpcService{
|
||||||
|
{
|
||||||
|
TargetSpecifier: &envoy_config_core_v3.GrpcService_EnvoyGrpc_{
|
||||||
|
EnvoyGrpc: &envoy_config_core_v3.GrpcService_EnvoyGrpc{
|
||||||
|
ClusterName: "pomerium-control-plane-grpc",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
LdsConfig: &envoy_config_core_v3.ConfigSource{
|
||||||
|
ResourceApiVersion: envoy_config_core_v3.ApiVersion_V3,
|
||||||
|
ConfigSourceSpecifier: &envoy_config_core_v3.ConfigSource_Ads{},
|
||||||
|
},
|
||||||
|
CdsConfig: &envoy_config_core_v3.ConfigSource{
|
||||||
|
ResourceApiVersion: envoy_config_core_v3.ApiVersion_V3,
|
||||||
|
ConfigSourceSpecifier: &envoy_config_core_v3.ConfigSource_Ads{},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
controlPlanePort, err := strconv.Atoi(srv.grpcPort)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("invalid control plane port: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
controlPlaneEndpoint := &envoy_config_endpoint_v3.LbEndpoint_Endpoint{
|
||||||
|
Endpoint: &envoy_config_endpoint_v3.Endpoint{
|
||||||
|
Address: &envoy_config_core_v3.Address{
|
||||||
|
Address: &envoy_config_core_v3.Address_SocketAddress{
|
||||||
|
SocketAddress: &envoy_config_core_v3.SocketAddress{
|
||||||
|
Address: "127.0.0.1",
|
||||||
|
PortSpecifier: &envoy_config_core_v3.SocketAddress_PortValue{
|
||||||
|
PortValue: uint32(controlPlanePort),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
controlPlaneCluster := &envoy_config_cluster_v3.Cluster{
|
||||||
|
Name: "pomerium-control-plane-grpc",
|
||||||
|
ConnectTimeout: &durationpb.Duration{
|
||||||
|
Seconds: 5,
|
||||||
|
},
|
||||||
|
ClusterDiscoveryType: &envoy_config_cluster_v3.Cluster_Type{
|
||||||
|
Type: envoy_config_cluster_v3.Cluster_STATIC,
|
||||||
|
},
|
||||||
|
LbPolicy: envoy_config_cluster_v3.Cluster_ROUND_ROBIN,
|
||||||
|
LoadAssignment: &envoy_config_endpoint_v3.ClusterLoadAssignment{
|
||||||
|
ClusterName: "pomerium-control-plane-grpc",
|
||||||
|
Endpoints: []*envoy_config_endpoint_v3.LocalityLbEndpoints{
|
||||||
|
{
|
||||||
|
LbEndpoints: []*envoy_config_endpoint_v3.LbEndpoint{
|
||||||
|
{
|
||||||
|
HostIdentifier: controlPlaneEndpoint,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
Http2ProtocolOptions: &envoy_config_core_v3.Http2ProtocolOptions{},
|
||||||
|
}
|
||||||
|
|
||||||
|
staticCfg := &envoy_config_bootstrap_v3.Bootstrap_StaticResources{
|
||||||
|
Clusters: []*envoy_config_cluster_v3.Cluster{
|
||||||
|
controlPlaneCluster,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg := &envoy_config_bootstrap_v3.Bootstrap{
|
||||||
|
Node: nodeCfg,
|
||||||
|
Admin: adminCfg,
|
||||||
|
DynamicResources: dynamicCfg,
|
||||||
|
StaticResources: staticCfg,
|
||||||
|
}
|
||||||
|
|
||||||
|
traceOpts, err := config.NewTracingOptions(srv.opts)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("invalid tracing config: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := srv.addTraceConfig(traceOpts, cfg); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to add tracing config: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
jsonBytes, err := protojson.Marshal(proto.MessageV2(cfg))
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return jsonBytes, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (srv *Server) addTraceConfig(traceOpts *config.TracingOptions, bootCfg *envoy_config_bootstrap_v3.Bootstrap) error {
|
||||||
|
|
||||||
|
if !traceOpts.Enabled() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
zipkinPort, err := strconv.Atoi(traceOpts.ZipkinEndpoint.Port())
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("invalid port number: %w", err)
|
||||||
|
}
|
||||||
|
zipkinAddress := traceOpts.ZipkinEndpoint.Hostname()
|
||||||
|
const zipkinClusterName = "zipkin"
|
||||||
|
|
||||||
|
zipkinEndpoint := &envoy_config_endpoint_v3.LbEndpoint_Endpoint{
|
||||||
|
Endpoint: &envoy_config_endpoint_v3.Endpoint{
|
||||||
|
Address: &envoy_config_core_v3.Address{
|
||||||
|
Address: &envoy_config_core_v3.Address_SocketAddress{
|
||||||
|
SocketAddress: &envoy_config_core_v3.SocketAddress{
|
||||||
|
Address: zipkinAddress,
|
||||||
|
PortSpecifier: &envoy_config_core_v3.SocketAddress_PortValue{
|
||||||
|
PortValue: uint32(zipkinPort),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
zipkinCluster := &envoy_config_cluster_v3.Cluster{
|
||||||
|
Name: zipkinClusterName,
|
||||||
|
ConnectTimeout: &durationpb.Duration{
|
||||||
|
Seconds: 10,
|
||||||
|
},
|
||||||
|
ClusterDiscoveryType: &envoy_config_cluster_v3.Cluster_Type{
|
||||||
|
Type: envoy_config_cluster_v3.Cluster_STATIC,
|
||||||
|
},
|
||||||
|
LbPolicy: envoy_config_cluster_v3.Cluster_ROUND_ROBIN,
|
||||||
|
LoadAssignment: &envoy_config_endpoint_v3.ClusterLoadAssignment{
|
||||||
|
ClusterName: zipkinClusterName,
|
||||||
|
Endpoints: []*envoy_config_endpoint_v3.LocalityLbEndpoints{
|
||||||
|
{
|
||||||
|
LbEndpoints: []*envoy_config_endpoint_v3.LbEndpoint{
|
||||||
|
{
|
||||||
|
HostIdentifier: zipkinEndpoint,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
tracingTC, _ := ptypes.MarshalAny(
|
||||||
|
&envoy_config_trace_v3.ZipkinConfig{
|
||||||
|
CollectorCluster: zipkinClusterName,
|
||||||
|
CollectorEndpoint: traceOpts.ZipkinEndpoint.Path,
|
||||||
|
CollectorEndpointVersion: envoy_config_trace_v3.ZipkinConfig_HTTP_JSON,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
tracingCfg := &envoy_config_trace_v3.Tracing{
|
||||||
|
Http: &envoy_config_trace_v3.Tracing_Http{
|
||||||
|
Name: "envoy.tracers.zipkin",
|
||||||
|
ConfigType: &envoy_config_trace_v3.Tracing_Http_TypedConfig{
|
||||||
|
TypedConfig: tracingTC,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
bootCfg.StaticResources.Clusters = append(bootCfg.StaticResources.Clusters, zipkinCluster)
|
||||||
|
bootCfg.Tracing = tracingCfg
|
||||||
|
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (srv *Server) handleLogs(stdout io.ReadCloser) {
|
func (srv *Server) handleLogs(stdout io.ReadCloser) {
|
||||||
|
|
|
@ -10,16 +10,12 @@ import (
|
||||||
prom "github.com/prometheus/client_golang/prometheus"
|
prom "github.com/prometheus/client_golang/prometheus"
|
||||||
"go.opencensus.io/stats/view"
|
"go.opencensus.io/stats/view"
|
||||||
|
|
||||||
"github.com/pomerium/pomerium/internal/envoy"
|
|
||||||
log "github.com/pomerium/pomerium/internal/log"
|
log "github.com/pomerium/pomerium/internal/log"
|
||||||
"github.com/pomerium/pomerium/internal/urlutil"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var envoyURL = envoy.EnvoyAdminURL
|
|
||||||
|
|
||||||
// PrometheusHandler creates an exporter that exports stats to Prometheus
|
// PrometheusHandler creates an exporter that exports stats to Prometheus
|
||||||
// and returns a handler suitable for exporting metrics.
|
// and returns a handler suitable for exporting metrics.
|
||||||
func PrometheusHandler() (http.Handler, error) {
|
func PrometheusHandler(envoyURL *url.URL) (http.Handler, error) {
|
||||||
if err := registerDefaultViews(); err != nil {
|
if err := registerDefaultViews(); err != nil {
|
||||||
return nil, fmt.Errorf("telemetry/metrics: failed registering views")
|
return nil, fmt.Errorf("telemetry/metrics: failed registering views")
|
||||||
}
|
}
|
||||||
|
@ -35,7 +31,7 @@ func PrometheusHandler() (http.Handler, error) {
|
||||||
view.RegisterExporter(exporter)
|
view.RegisterExporter(exporter)
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
|
|
||||||
envoyMetricsURL, err := urlutil.ParseAndValidateURL(fmt.Sprintf("%s/stats/prometheus", envoyURL))
|
envoyMetricsURL, err := envoyURL.Parse("/stats/prometheus")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("telemetry/metrics: invalid proxy URL: %w", err)
|
return nil, fmt.Errorf("telemetry/metrics: invalid proxy URL: %w", err)
|
||||||
}
|
}
|
||||||
|
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
"regexp"
|
"regexp"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
|
@ -27,8 +28,8 @@ envoy_server_initialization_time_ms_bucket{le="1000"} 1
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func getMetrics(t *testing.T) []byte {
|
func getMetrics(t *testing.T, envoyURL *url.URL) []byte {
|
||||||
h, err := PrometheusHandler()
|
h, err := PrometheusHandler(envoyURL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
@ -48,7 +49,7 @@ func getMetrics(t *testing.T) []byte {
|
||||||
func Test_PrometheusHandler(t *testing.T) {
|
func Test_PrometheusHandler(t *testing.T) {
|
||||||
|
|
||||||
t.Run("no envoy", func(t *testing.T) {
|
t.Run("no envoy", func(t *testing.T) {
|
||||||
b := getMetrics(t)
|
b := getMetrics(t, &url.URL{})
|
||||||
|
|
||||||
if m, _ := regexp.Match(`(?m)^# HELP .*`, b); !m {
|
if m, _ := regexp.Match(`(?m)^# HELP .*`, b); !m {
|
||||||
t.Errorf("Metrics endpoint did not contain any help messages: %s", b)
|
t.Errorf("Metrics endpoint did not contain any help messages: %s", b)
|
||||||
|
@ -57,8 +58,8 @@ func Test_PrometheusHandler(t *testing.T) {
|
||||||
|
|
||||||
t.Run("with envoy", func(t *testing.T) {
|
t.Run("with envoy", func(t *testing.T) {
|
||||||
fakeEnvoyMetricsServer := httptest.NewServer(newEnvoyMetricsHandler())
|
fakeEnvoyMetricsServer := httptest.NewServer(newEnvoyMetricsHandler())
|
||||||
envoyURL = fakeEnvoyMetricsServer.URL
|
envoyURL, _ := url.Parse(fakeEnvoyMetricsServer.URL)
|
||||||
b := getMetrics(t)
|
b := getMetrics(t, envoyURL)
|
||||||
|
|
||||||
if m, _ := regexp.Match(`(?m)^go_.*`, b); !m {
|
if m, _ := regexp.Match(`(?m)^go_.*`, b); !m {
|
||||||
t.Errorf("Metrics endpoint did not contain internal metrics: %s", b)
|
t.Errorf("Metrics endpoint did not contain internal metrics: %s", b)
|
||||||
|
|
|
@ -10,47 +10,18 @@ import (
|
||||||
zipkinHTTP "github.com/openzipkin/zipkin-go/reporter/http"
|
zipkinHTTP "github.com/openzipkin/zipkin-go/reporter/http"
|
||||||
"go.opencensus.io/trace"
|
"go.opencensus.io/trace"
|
||||||
|
|
||||||
|
"github.com/pomerium/pomerium/config"
|
||||||
"github.com/pomerium/pomerium/internal/log"
|
"github.com/pomerium/pomerium/internal/log"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
|
||||||
// JaegerTracingProviderName is the name of the tracing provider Jaeger.
|
|
||||||
JaegerTracingProviderName = "jaeger"
|
|
||||||
// ZipkinTracingProviderName is the name of the tracing provider Zipkin.
|
|
||||||
ZipkinTracingProviderName = "zipkin"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TracingOptions contains the configurations settings for a http server.
|
|
||||||
type TracingOptions struct {
|
|
||||||
// Shared
|
|
||||||
Provider string
|
|
||||||
Service string
|
|
||||||
Debug bool
|
|
||||||
|
|
||||||
// Jaeger
|
|
||||||
|
|
||||||
// CollectorEndpoint is the full url to the Jaeger HTTP Thrift collector.
|
|
||||||
// For example, http://localhost:14268/api/traces
|
|
||||||
JaegerCollectorEndpoint string `mapstructure:"tracing_jaeger_collector_endpoint"`
|
|
||||||
// AgentEndpoint instructs exporter to send spans to jaeger-agent at this address.
|
|
||||||
// For example, localhost:6831.
|
|
||||||
JaegerAgentEndpoint string `mapstructure:"tracing_jaeger_agent_endpoint"`
|
|
||||||
|
|
||||||
// Zipkin
|
|
||||||
|
|
||||||
// ZipkinEndpoint configures the zipkin collector URI
|
|
||||||
// Example: http://zipkin:9411/api/v2/spans
|
|
||||||
ZipkinEndpoint string `mapstructure:"tracing_zipkin_endpoint"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// RegisterTracing creates a new trace exporter from TracingOptions.
|
// RegisterTracing creates a new trace exporter from TracingOptions.
|
||||||
func RegisterTracing(opts *TracingOptions) (trace.Exporter, error) {
|
func RegisterTracing(opts *config.TracingOptions) (trace.Exporter, error) {
|
||||||
var exporter trace.Exporter
|
var exporter trace.Exporter
|
||||||
var err error
|
var err error
|
||||||
switch opts.Provider {
|
switch opts.Provider {
|
||||||
case JaegerTracingProviderName:
|
case config.JaegerTracingProviderName:
|
||||||
exporter, err = registerJaeger(opts)
|
exporter, err = registerJaeger(opts)
|
||||||
case ZipkinTracingProviderName:
|
case config.ZipkinTracingProviderName:
|
||||||
exporter, err = registerZipkin(opts)
|
exporter, err = registerZipkin(opts)
|
||||||
default:
|
default:
|
||||||
return nil, fmt.Errorf("telemetry/trace: provider %s unknown", opts.Provider)
|
return nil, fmt.Errorf("telemetry/trace: provider %s unknown", opts.Provider)
|
||||||
|
@ -58,10 +29,8 @@ func RegisterTracing(opts *TracingOptions) (trace.Exporter, error) {
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if opts.Debug {
|
trace.ApplyConfig(trace.Config{DefaultSampler: trace.ProbabilitySampler(opts.SampleRate)})
|
||||||
log.Debug().Msg("telemetry/trace: debug on, sample everything")
|
|
||||||
trace.ApplyConfig(trace.Config{DefaultSampler: trace.AlwaysSample()})
|
|
||||||
}
|
|
||||||
log.Debug().Interface("Opts", opts).Msg("telemetry/trace: exporter created")
|
log.Debug().Interface("Opts", opts).Msg("telemetry/trace: exporter created")
|
||||||
return exporter, nil
|
return exporter, nil
|
||||||
}
|
}
|
||||||
|
@ -71,13 +40,15 @@ func UnregisterTracing(exporter trace.Exporter) {
|
||||||
trace.UnregisterExporter(exporter)
|
trace.UnregisterExporter(exporter)
|
||||||
}
|
}
|
||||||
|
|
||||||
func registerJaeger(opts *TracingOptions) (trace.Exporter, error) {
|
func registerJaeger(opts *config.TracingOptions) (trace.Exporter, error) {
|
||||||
jex, err := jaeger.NewExporter(
|
jOpts := jaeger.Options{
|
||||||
jaeger.Options{
|
|
||||||
AgentEndpoint: opts.JaegerAgentEndpoint,
|
|
||||||
CollectorEndpoint: opts.JaegerCollectorEndpoint,
|
|
||||||
ServiceName: opts.Service,
|
ServiceName: opts.Service,
|
||||||
})
|
AgentEndpoint: opts.JaegerAgentEndpoint,
|
||||||
|
}
|
||||||
|
if opts.JaegerCollectorEndpoint != nil {
|
||||||
|
jOpts.CollectorEndpoint = opts.JaegerCollectorEndpoint.String()
|
||||||
|
}
|
||||||
|
jex, err := jaeger.NewExporter(jOpts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -85,13 +56,13 @@ func registerJaeger(opts *TracingOptions) (trace.Exporter, error) {
|
||||||
return jex, nil
|
return jex, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func registerZipkin(opts *TracingOptions) (trace.Exporter, error) {
|
func registerZipkin(opts *config.TracingOptions) (trace.Exporter, error) {
|
||||||
localEndpoint, err := zipkin.NewEndpoint(opts.Service, "")
|
localEndpoint, err := zipkin.NewEndpoint(opts.Service, "")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("telemetry/trace: could not create local endpoint: %w", err)
|
return nil, fmt.Errorf("telemetry/trace: could not create local endpoint: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
reporter := zipkinHTTP.NewReporter(opts.ZipkinEndpoint)
|
reporter := zipkinHTTP.NewReporter(opts.ZipkinEndpoint.String())
|
||||||
|
|
||||||
exporter := ocZipkin.NewExporter(reporter, localEndpoint)
|
exporter := ocZipkin.NewExporter(reporter, localEndpoint)
|
||||||
trace.RegisterExporter(exporter)
|
trace.RegisterExporter(exporter)
|
||||||
|
|
|
@ -1,17 +1,23 @@
|
||||||
package trace
|
package trace
|
||||||
|
|
||||||
import "testing"
|
import (
|
||||||
|
"net/url"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/pomerium/pomerium/config"
|
||||||
|
)
|
||||||
|
|
||||||
func TestRegisterTracing(t *testing.T) {
|
func TestRegisterTracing(t *testing.T) {
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
opts *TracingOptions
|
opts *config.TracingOptions
|
||||||
wantErr bool
|
wantErr bool
|
||||||
}{
|
}{
|
||||||
{"jaeger", &TracingOptions{JaegerAgentEndpoint: "localhost:6831", Service: "all", Provider: "jaeger"}, false},
|
{"jaeger", &config.TracingOptions{JaegerAgentEndpoint: "localhost:6831", Service: "all", Provider: "jaeger"}, false},
|
||||||
{"jaeger with debug", &TracingOptions{JaegerAgentEndpoint: "localhost:6831", Service: "all", Provider: "jaeger", Debug: true}, false},
|
{"jaeger with debug", &config.TracingOptions{JaegerAgentEndpoint: "localhost:6831", Service: "all", Provider: "jaeger", Debug: true}, false},
|
||||||
{"jaeger no endpoint", &TracingOptions{JaegerAgentEndpoint: "", Service: "all", Provider: "jaeger"}, true},
|
{"jaeger no endpoint", &config.TracingOptions{JaegerAgentEndpoint: "", Service: "all", Provider: "jaeger"}, true},
|
||||||
{"unknown provider", &TracingOptions{JaegerAgentEndpoint: "localhost:0", Service: "all", Provider: "Lucius Cornelius Sulla"}, true},
|
{"unknown provider", &config.TracingOptions{JaegerAgentEndpoint: "localhost:0", Service: "all", Provider: "Lucius Cornelius Sulla"}, true},
|
||||||
|
{"zipkin with debug", &config.TracingOptions{ZipkinEndpoint: &url.URL{Host: "localhost"}, Service: "all", Provider: "zipkin", Debug: true}, false},
|
||||||
}
|
}
|
||||||
for _, tt := range tests {
|
for _, tt := range tests {
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue