Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions cobra.go
Original file line number Diff line number Diff line change
Expand Up @@ -114,13 +114,13 @@ func NewCobraOpts(cmd *cobra.Command, opts *TemporalOpts) *TemporalOpts {
viper.SetDefault("health_listen_address", "0.0.0.0:3000")
cmd.Flags().StringVar(
&opts.HealthListenAddress, "health-listen-address",
viper.GetString("health_listen_address"), "Address of health server",
viper.GetString("health_listen_address"), "Address of health server. If the port is 0, a random available port is assigned.",
)

viper.SetDefault("metrics_listen_address", "0.0.0.0:9090")
cmd.Flags().StringVar(
&opts.MetricsListenAddress, "metrics-listen-address",
viper.GetString("metrics_listen_address"), "Address of Prometheus metrics server",
viper.GetString("metrics_listen_address"), "Address of Prometheus metrics server. If the port is 0, a random available port is assigned.",
)

cmd.Flags().StringVar(
Expand Down
10 changes: 5 additions & 5 deletions health.go
Original file line number Diff line number Diff line change
Expand Up @@ -229,16 +229,16 @@ func NewHealthCheck(ctx context.Context, taskQueues []string, address string, c
Handler: mux,
}

l := log.With().
Str("address", address).
Int("taskQueueCount", len(taskQueues)).
Logger()

listener, err := (&net.ListenConfig{}).Listen(ctx, "tcp", address)
if err != nil {
return fmt.Errorf("error listening on %s: %w", address, err)
}

l := log.With().
Str("address", listener.Addr().String()).
Int("taskQueueCount", len(taskQueues)).
Logger()

go func() {
<-ctx.Done()

Expand Down
26 changes: 26 additions & 0 deletions health_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -683,6 +683,32 @@ func TestNewHealthCheck(t *testing.T) {
assert.JSONEq(t, `{"healthy":true}`, body, "the fake client reports a healthy Temporal")
})

t.Run("port 0 serves on an assigned port and logs it", func(t *testing.T) {
logs := captureLogs(t)

ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)

require.NoError(t, NewHealthCheck(ctx, []string{testTaskQueue}, "127.0.0.1:0", newFakeTemporalClient()))

// NewHealthCheck does not return the address, so the log is the only
// way to find the port the OS assigned.
entry := logs.waitForMessage(t, "Starting healthcheck service")
address := requireAssignedAddress(t, entry, "127.0.0.1")
assert.InDelta(t, 1, entry["taskQueueCount"], 0)

httpClient := newTestHTTPClient(t)

status, body, err := getLiveness(t, httpClient, address)
require.NoError(t, err, "the healthcheck serves on the logged address")
assert.Equal(t, http.StatusOK, status)
assert.JSONEq(t, `{"healthy":true}`, body)

cancel()

requireNotServing(t, httpClient, address)
})

t.Run("shuts down when the context is cancelled", func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Expand Down
107 changes: 107 additions & 0 deletions helpers_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,20 @@ import (
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/json"
"encoding/pem"
"math/big"
"net"
"os"
"path/filepath"
"reflect"
"strconv"
"sync"
"testing"
"time"

"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.temporal.io/sdk/client"
Expand Down Expand Up @@ -154,3 +160,104 @@ func writeFile(t *testing.T, name, content string) string {

return path
}

// logWaitTimeout bounds how long a test waits for a log entry written from
// another goroutine. It only exists so a broken implementation fails rather than
// hangs.
const logWaitTimeout = 5 * time.Second

// logCapture records the JSON entries written by a zerolog logger. It is mutex
// guarded because servers log from their own goroutines.
type logCapture struct {
mu sync.Mutex
entries []map[string]any
// written is signalled after every entry, so waiters never poll.
written chan struct{}
}

func (c *logCapture) Write(p []byte) (int, error) {
entry := map[string]any{}
if err := json.Unmarshal(p, &entry); err != nil {
return 0, err
}

c.mu.Lock()
c.entries = append(c.entries, entry)
c.mu.Unlock()

select {
case c.written <- struct{}{}:
default:
}

return len(p), nil
}

func (c *logCapture) find(message string) (map[string]any, bool) {
c.mu.Lock()
defer c.mu.Unlock()

for _, entry := range c.entries {
if entry[zerolog.MessageFieldName] == message {
return entry, true
}
}

return nil, false
}

// waitForMessage returns the first entry with the given message, waiting a
// bounded time for it to be written.
func (c *logCapture) waitForMessage(t *testing.T, message string) map[string]any {
t.Helper()

timeout := time.NewTimer(logWaitTimeout)
defer timeout.Stop()

for {
if entry, ok := c.find(message); ok {
return entry
}

select {
case <-c.written:
case <-timeout.C:
t.Fatalf("no %q log entry after %s", message, logWaitTimeout)
}
}
}

// captureLogs swaps the global zerolog logger for one that records its entries
// and restores the original when the test finishes. Loggers derived from the
// global one before the swap is undone keep writing to the capture, so entries
// logged from a goroutine are still recorded. Callers must not be parallel.
func captureLogs(t *testing.T) *logCapture {
t.Helper()

original := log.Logger
t.Cleanup(func() { log.Logger = original })

c := &logCapture{written: make(chan struct{}, 1)}
log.Logger = zerolog.New(c)

return c
}

// requireAssignedAddress asserts that a logged address is on the expected host
// with a real, OS-assigned port rather than 0, and returns it.
func requireAssignedAddress(t *testing.T, entry map[string]any, expectedHost string) string {
t.Helper()

address, ok := entry["address"].(string)
require.True(t, ok, "the log entry has an address: %v", entry)

host, port, err := net.SplitHostPort(address)
require.NoError(t, err)
assert.Equal(t, expectedHost, host)

p, err := strconv.Atoi(port)
require.NoError(t, err)
require.NotZero(t, p, "the logged port is the assigned one, not 0")

return address
}
116 changes: 87 additions & 29 deletions prometheus.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,12 @@
package temporal

import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"time"

prom "github.com/prometheus/client_golang/prometheus"
Expand All @@ -37,28 +41,51 @@ import (
// handler, and is responsible for calling [PrometheusHandler.Close].
type PrometheusHandler struct {
client.MetricsHandler
closer io.Closer
closer io.Closer
server *http.Server
listener net.Listener
}

// Close flushes and releases the underlying Tally scope, and reports whatever
// the scope reports. A handler with no scope to release, such as the zero
// value, closes without error.
// Close stops the owned HTTP server and listener, then flushes and releases the
// underlying Tally scope. It returns any errors encountered during cleanup.
// The zero value closes without error. An endpoint registered on Go's default
// HTTP mux remains registered.
func (h *PrometheusHandler) Close() error {
if h.closer == nil {
return nil
var errs []error

if h.server != nil {
if err := h.server.Close(); err != nil {
errs = append(errs, err)
}
}

// Also close the listener directly in case Close was called before
// the Serve goroutine registered it with the HTTP server.
if h.listener != nil {
if err := h.listener.Close(); err != nil && !errors.Is(err, net.ErrClosed) {
errs = append(errs, err)
}
}

if h.closer != nil {
if err := h.closer.Close(); err != nil {
errs = append(errs, err)
}
}

return h.closer.Close()
return errors.Join(errs...)
}

// NewPrometheusHandler creates a Prometheus-backed metrics handler for a
// Temporal client. The caller owns it and should call
// [PrometheusHandler.Close] when it is no longer needed, usually with defer.
//
// Metrics are served on listenAddress under /metrics, with prefix prepended to
// every metric name. An empty listenAddress attaches the metrics endpoint to
// Go's default HTTP mux instead of starting a server of its own. A nil registry
// uses the Prometheus default registry. Timers are reported as histograms.
// every metric name. A port of 0 (for example "0.0.0.0:0") lets the OS pick a
// free port, which is logged. An empty listenAddress attaches the metrics
// endpoint to Go's default HTTP mux instead of starting a server of its own. A
// nil registry uses the Prometheus default registry. Timers are reported as
// histograms.
//
// The optional onError argument decides how the reporter reports its own
// failures:
Expand All @@ -69,20 +96,14 @@ func (h *PrometheusHandler) Close() error {
// behaviour applies
// - more than one: an error is returned and no handler is created
//
// Reporter failures are not returned from here. A listen address that cannot be
// bound, for example, is reported asynchronously through onError once the
// reporter is running.
// Listen failures are returned synchronously. Metric registration and unexpected
// HTTP serving failures are reported through onError.
func NewPrometheusHandler(
listenAddress,
prefix string,
registry *prom.Registry,
onError ...func(error),
) (*PrometheusHandler, error) {
c := prometheus.Configuration{
ListenAddress: listenAddress,
TimerType: "histogram",
}

errorHandler := func(err error) {
log.Fatal().Err(err).Msg("Error in Prometheus reporter")
}
Expand All @@ -93,14 +114,35 @@ func NewPrometheusHandler(
errorHandler = onError[0]
}

reporter, err := c.NewReporter(
prometheus.ConfigurationOptions{
Registry: registry,
OnError: errorHandler,
},
)
if err != nil {
return nil, fmt.Errorf("error creating reporter: %w", err)
reporterOpts := prometheus.Options{
DefaultTimerType: prometheus.HistogramTimerType,
OnRegisterError: errorHandler,
}
if registry != nil {
reporterOpts.Registerer = registry
reporterOpts.Gatherer = registry
}

reporter := prometheus.NewReporter(reporterOpts)

var listener net.Listener
var server *http.Server

if listenAddress == "" {
http.Handle("/metrics", reporter.HTTPHandler())
} else {
var err error
listener, err = (&net.ListenConfig{}).Listen(context.Background(), "tcp", listenAddress)
if err != nil {
return nil, fmt.Errorf("error listening for Prometheus: %w", err)
}

mux := http.NewServeMux()
mux.Handle("/metrics", reporter.HTTPHandler())
server = &http.Server{
Handler: mux,
ReadHeaderTimeout: 5 * time.Second,
}
}

scopeOpts := tally.ScopeOptions{
Expand All @@ -112,9 +154,25 @@ func NewPrometheusHandler(
scope, closer := tally.NewRootScope(scopeOpts, time.Second)
scope = sdktally.NewPrometheusNamingScope(scope)

log.Info().Str("address", listenAddress).Msg("Starting Prometheus service")
return &PrometheusHandler{
handler := &PrometheusHandler{
MetricsHandler: sdktally.NewMetricsHandler(scope),
closer: closer,
}, nil
server: server,
listener: listener,
}

if server != nil {
log.Info().Str("address", listener.Addr().String()).Msg("Starting Prometheus service")

go func() {
if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
if errorHandler == nil {
panic(err)
}
errorHandler(err)
}
}()
}

return handler, nil
}
Loading
Loading