Merge pull request 'meter: move metrics handling in broker implementations' (#215) from metrics into master

Reviewed-on: #215

broker/memory.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"context"
 	"sync"
+	"time"
 
 	"go.unistack.org/micro/v4/logger"
 	"go.unistack.org/micro/v4/metadata"
@@ -134,13 +135,22 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*Message, opts ...Pub
 		eh := m.opts.ErrorHandler
 
 		for t, ms := range msgTopicMap {
+			ts := time.Now()
+
+			m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(len(ms))
+			m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(len(ms))
+
 			m.RLock()
 			subs, ok := m.subscribers[t]
 			m.RUnlock()
 			if !ok {
+				m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "failure").Add(len(ms))
+				m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-len(ms))
+				m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
 				continue
 			}
 
+			m.opts.Meter.Counter(PublishMessageTotal, "endpoint", t, "status", "success").Add(len(ms))
 			for _, sub := range subs {
 				if sub.opts.BatchErrorHandler != nil {
 					beh = sub.opts.BatchErrorHandler
@@ -152,37 +162,65 @@ func (m *memoryBroker) publish(ctx context.Context, msgs []*Message, opts ...Pub
 				switch {
 				// batch processing
 				case sub.batchhandler != nil:
+
 					if err = sub.batchhandler(ms); err != nil {
+						m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
 						ms.SetError(err)
 						if beh != nil {
 							_ = beh(ms)
 						} else if m.opts.Logger.V(logger.ErrorLevel) {
 							m.opts.Logger.Error(m.opts.Context, err.Error())
 						}
-					} else if sub.opts.AutoAck {
+					} else {
+						if sub.opts.AutoAck {
 							if err = ms.Ack(); err != nil {
 								m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+								m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
+							} else {
+								m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+							}
+						} else {
+							m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
 						}
 					}
+					m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-len(ms))
+					m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-len(ms))
 					// single processing
 				case sub.handler != nil:
 					for _, p := range ms {
 						if err = sub.handler(p); err != nil {
+							m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
 							p.SetError(err)
 							if eh != nil {
 								_ = eh(p)
 							} else if m.opts.Logger.V(logger.ErrorLevel) {
 								m.opts.Logger.Error(m.opts.Context, err.Error())
 							}
-						} else if sub.opts.AutoAck {
+						} else {
+							if sub.opts.AutoAck {
 								if err = p.Ack(); err != nil {
 									m.opts.Logger.Errorf(m.opts.Context, "ack failed: %v", err)
+									m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "failure").Inc()
+								} else {
+									m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
+								}
+							} else {
+								m.opts.Meter.Counter(SubscribeMessageTotal, "endpoint", t, "status", "success").Inc()
 							}
 						}
+						m.opts.Meter.Counter(PublishMessageInflight, "endpoint", t).Add(-1)
+						m.opts.Meter.Counter(SubscribeMessageInflight, "endpoint", t).Add(-1)
 					}
 				}
+
 			}
+			te := time.Since(ts)
+			m.opts.Meter.Summary(PublishMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Histogram(PublishMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, "endpoint", t).Update(te.Seconds())
+			m.opts.Meter.Histogram(SubscribeMessageDurationSeconds, "endpoint", t).Update(te.Seconds())
 		}
+
 	}
 
 	return nil

broker/options.go

@@ -12,6 +12,25 @@ import (
 	"go.unistack.org/micro/v4/tracer"
 )
 
+var (
+	// PublishMessageDurationSeconds specifies meter metric name
+	PublishMessageDurationSeconds = "publish_message_duration_seconds"
+	// PublishMessageLatencyMicroseconds specifies meter metric name
+	PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
+	// PublishMessageTotal specifies meter metric name
+	PublishMessageTotal = "publish_message_total"
+	// PublishMessageInflight specifies meter metric name
+	PublishMessageInflight = "publish_message_inflight"
+	// SubscribeMessageDurationSeconds specifies meter metric name
+	SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
+	// SubscribeMessageLatencyMicroseconds specifies meter metric name
+	SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
+	// SubscribeMessageTotal specifies meter metric name
+	SubscribeMessageTotal = "subscribe_message_total"
+	// SubscribeMessageInflight specifies meter metric name
+	SubscribeMessageInflight = "subscribe_message_inflight"
+)
+
 // Options struct
 type Options struct {
 	// Tracer used for tracing

meter/wrapper/wrapper.go

@@ -27,22 +27,6 @@ var (
 	ServerRequestTotal = "server_request_total"
 	// ServerRequestInflight specifies meter metric name
 	ServerRequestInflight = "server_request_inflight"
-	// PublishMessageDurationSeconds specifies meter metric name
-	PublishMessageDurationSeconds = "publish_message_duration_seconds"
-	// PublishMessageLatencyMicroseconds specifies meter metric name
-	PublishMessageLatencyMicroseconds = "publish_message_latency_microseconds"
-	// PublishMessageTotal specifies meter metric name
-	PublishMessageTotal = "publish_message_total"
-	// PublishMessageInflight specifies meter metric name
-	PublishMessageInflight = "publish_message_inflight"
-	// SubscribeMessageDurationSeconds specifies meter metric name
-	SubscribeMessageDurationSeconds = "subscribe_message_duration_seconds"
-	// SubscribeMessageLatencyMicroseconds specifies meter metric name
-	SubscribeMessageLatencyMicroseconds = "subscribe_message_latency_microseconds"
-	// SubscribeMessageTotal specifies meter metric name
-	SubscribeMessageTotal = "subscribe_message_total"
-	// SubscribeMessageInflight specifies meter metric name
-	SubscribeMessageInflight = "subscribe_message_inflight"
 
 	labelSuccess  = "success"
 	labelFailure  = "failure"
@@ -230,37 +214,7 @@ func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client
 }
 
 func (w *wrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
-	endpoint := p.Topic()
+	return w.Client.Publish(ctx, p, opts...)
-
-	labels := make([]string, 0, 4)
-	labels = append(labels, labelEndpoint, endpoint)
-
-	w.opts.Meter.Counter(PublishMessageInflight, labels...).Inc()
-	ts := time.Now()
-	err := w.Client.Publish(ctx, p, opts...)
-	te := time.Since(ts)
-	w.opts.Meter.Counter(PublishMessageInflight, labels...).Dec()
-
-	w.opts.Meter.Summary(PublishMessageLatencyMicroseconds, labels...).Update(te.Seconds())
-	w.opts.Meter.Histogram(PublishMessageDurationSeconds, labels...).Update(te.Seconds())
-
-	if err == nil {
-		labels = append(labels, labelStatus, labelSuccess)
-	} else {
-		labels = append(labels, labelStatus, labelFailure)
-	}
-	w.opts.Meter.Counter(PublishMessageTotal, labels...).Inc()
-
-	return err
-}
-
-// NewHandlerWrapper create new server handler wrapper
-// deprecated
-func NewHandlerWrapper(opts ...Option) server.HandlerWrapper {
-	handler := &wrapper{
-		opts: NewOptions(opts...),
-	}
-	return handler.HandlerFunc
 }
 
 // NewServerHandlerWrapper create new server handler wrapper
@@ -302,46 +256,3 @@ func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
 		return err
 	}
 }
-
-// NewSubscriberWrapper create server subscribe wrapper
-// deprecated
-func NewSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
-	handler := &wrapper{
-		opts: NewOptions(opts...),
-	}
-	return handler.SubscriberFunc
-}
-
-func NewServerSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
-	handler := &wrapper{
-		opts: NewOptions(opts...),
-	}
-	return handler.SubscriberFunc
-}
-
-func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc {
-	return func(ctx context.Context, msg server.Message) error {
-		endpoint := msg.Topic()
-
-		labels := make([]string, 0, 4)
-		labels = append(labels, labelEndpoint, endpoint)
-
-		w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Inc()
-		ts := time.Now()
-		err := fn(ctx, msg)
-		te := time.Since(ts)
-		w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Dec()
-
-		w.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, labels...).Update(te.Seconds())
-		w.opts.Meter.Histogram(SubscribeMessageDurationSeconds, labels...).Update(te.Seconds())
-
-		if err == nil {
-			labels = append(labels, labelStatus, labelSuccess)
-		} else {
-			labels = append(labels, labelStatus, labelFailure)
-		}
-		w.opts.Meter.Counter(SubscribeMessageTotal, labels...).Inc()
-
-		return err
-	}
-}

util/io/redirect.go

@@ -0,0 +1,17 @@
+package io
+
+import (
+	"os"
+	"sync"
+)
+
+var osStderrMu sync.Mutex
+
+var OrigStderr = func() *os.File {
+	fd, err := dupFD(os.Stderr.Fd())
+	if err != nil {
+		panic(err)
+	}
+
+	return os.NewFile(fd, os.Stderr.Name())
+}()

util/io/redirect_test.go

@@ -0,0 +1,40 @@
+package io
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"os"
+	"regexp"
+	"testing"
+	"time"
+)
+
+var ErrorPattern = regexp.MustCompile(`"error"`)
+
+func TestRedirect(t *testing.T) {
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch := make(chan string)
+
+	go func() {
+		buf := bytes.NewBuffer(nil)
+		_, _ = io.Copy(buf, r)
+		ch <- buf.String()
+	}()
+
+	if err = RedirectStderr(w); err != nil {
+		t.Fatal(err)
+	}
+
+	os.Stderr.Write([]byte(`test redirect`))
+	time.Sleep(1 * time.Millisecond)
+	r.Close()
+	str := <-ch
+	if ErrorPattern.MatchString(str) {
+		t.Fatal(fmt.Errorf(str))
+	}
+}

util/io/redirect_unix.go

@@ -0,0 +1,48 @@
+//go:build !windows
+// +build !windows
+
+package io
+
+import (
+	"os"
+
+	"golang.org/x/sys/unix"
+)
+
+// dupFD is used to initialize OrigStderr (see stderr_redirect.go).
+func dupFD(fd uintptr) (uintptr, error) {
+	// Warning: failing to set FD_CLOEXEC causes the duplicated file descriptor
+	// to leak into subprocesses created by exec.Command. If the file descriptor
+	// is a pipe, these subprocesses will hold the pipe open (i.e., prevent
+	// EOF), potentially beyond the lifetime of this process.
+	//
+	// This can break go test's timeouts. go test usually spawns a test process
+	// with its stdin and stderr streams hooked up to pipes; if the test process
+	// times out, it sends a SIGKILL and attempts to read stdin and stderr to
+	// completion. If the test process has itself spawned long-lived
+	// subprocesses that hold references to the stdin or stderr pipes, go test
+	// will hang until the subprocesses exit, rather defeating the purpose of
+	// a timeout.
+	nfd, err := unix.FcntlInt(fd, unix.F_DUPFD_CLOEXEC, 0)
+	if err != nil {
+		return 0, err
+	}
+	return uintptr(nfd), nil
+}
+
+// RedirectStderr is used to redirect internal writes to fd 2 to the
+// specified file. This is needed to ensure that harcoded writes to fd
+// 2 by e.g. the Go runtime are redirected to a log file of our
+// choosing.
+//
+// We also override os.Stderr for those other parts of Go which use
+// that and not fd 2 directly.
+func RedirectStderr(f *os.File) error {
+	osStderrMu.Lock()
+	defer osStderrMu.Unlock()
+	if err := unix.Dup2(int(f.Fd()), unix.Stderr); err != nil {
+		return err
+	}
+	os.Stderr = f
+	return nil
+}

util/io/redirect_windows.go

@@ -0,0 +1,32 @@
+package io
+
+import (
+	"os"
+
+	"golang.org/x/sys/windows"
+)
+
+// dupFD is used to initialize OrigStderr (see stderr_redirect.go).
+func dupFD(fd uintptr) (uintptr, error) {
+	// Adapted from https://github.com/golang/go/blob/go1.8/src/syscall/exec_windows.go#L303.
+	p := windows.CurrentProcess()
+	var h windows.Handle
+	return uintptr(h), windows.DuplicateHandle(p, windows.Handle(fd), p, &h, 0, true, windows.DUPLICATE_SAME_ACCESS)
+}
+
+// RedirectStderr is used to redirect internal writes to the error
+// handle to the specified file. This is needed to ensure that
+// harcoded writes to the error handle by e.g. the Go runtime are
+// redirected to a log file of our choosing.
+//
+// We also override os.Stderr for those other parts of Go which use
+// that and not fd 2 directly.
+func RedirectStderr(f *os.File) error {
+	osStderrMu.Lock()
+	defer osStderrMu.Unlock()
+	if err := windows.SetStdHandle(windows.STD_ERROR_HANDLE, windows.Handle(f.Fd())); err != nil {
+		return err
+	}
+	os.Stderr = f
+	return nil
+}

util/test/test.go

@@ -157,7 +157,7 @@ func SQLFromReader(m sqlmock.Sqlmock, r io.Reader) error {
 }
 
 func RunWithClientExpectResults(ctx context.Context, c client.Client, m sqlmock.Sqlmock, dir string, exts []string) error {
-	tcases, err := getFiles(dir, exts)
+	tcases, err := GetCases(dir, exts)
 	if err != nil {
 		return err
 	}
@@ -218,7 +218,7 @@ type Case struct {
 	rspfiles []string
 }
 
-func getFiles(dir string, exts []string) ([]Case, error) {
+func GetCases(dir string, exts []string) ([]Case, error) {
 	var tcases []Case
 	entries, err := os.ReadDir(dir)
 	if len(entries) == 0 && err != nil {
@@ -263,7 +263,7 @@ func getFiles(dir string, exts []string) ([]Case, error) {
 	}
 
 	for _, dir = range dirs {
-		ntcases, err := getFiles(dir, exts)
+		ntcases, err := GetCases(dir, exts)
 		if len(ntcases) == 0 && err != nil {
 			return tcases, err
 		} else if len(ntcases) == 0 {

util/test/test_test.go

@@ -56,8 +56,8 @@ func Test_SQLFromFile(t *testing.T) {
 	}
 }
 
-func Test_getFiles(t *testing.T) {
+func Test_GetCases(t *testing.T) {
-	files, err := getFiles("testdata/", nil)
+	files, err := GetCases("testdata/", nil)
 	if err != nil {
 		t.Fatal(err)
 	}