chore(tests): use a coverage server binary (#8407)

# Which Problems Are Solved

Use a single server instance for API integration tests. This optimizes
the time taken for the integration test pipeline,
because it allows running tests on multiple packages in parallel. Also,
it saves time by not start and stopping a zitadel server for every
package.

# How the Problems Are Solved

- Build a binary with `go build -race -cover ....`
- Integration tests only construct clients. The server remains running
in the background.
- The integration package and tested packages now fully utilize the API.
No more direct database access trough `query` and `command` packages.
- Use Makefile recipes to setup, start and stop the server in the
background.
- The binary has the race detector enabled
- Init and setup jobs are configured to halt immediately on race
condition
- Because the server runs in the background, races are only logged. When
the server is stopped and race logs exist, the Makefile recipe will
throw an error and print the logs.
- Makefile recipes include logic to print logs and convert coverage
reports after the server is stopped.
- Some tests need a downstream HTTP server to make requests, like quota
and milestones. A new `integration/sink` package creates an HTTP server
and uses websockets to forward HTTP request back to the test packages.
The package API uses Go channels for abstraction and easy usage.

# Additional Changes

- Integration test files already used the `//go:build integration`
directive. In order to properly split integration from unit tests,
integration test files need to be in a `integration_test` subdirectory
of their package.
- `UseIsolatedInstance` used to overwrite the `Tester.Client` for each
instance. Now a `Instance` object is returned with a gRPC client that is
connected to the isolated instance's hostname.
- The `Tester` type is now `Instance`. The object is created for the
first instance, used by default in any test. Isolated instances are also
`Instance` objects and therefore benefit from the same methods and
values. The first instance and any other us capable of creating an
isolated instance over the system API.
- All test packages run in an Isolated instance by calling
`NewInstance()`
- Individual tests that use an isolated instance use `t.Parallel()`

# Additional Context

- Closes #6684
- https://go.dev/doc/articles/race_detector
- https://go.dev/doc/build-cover

---------

Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com>

internal/integration/sink/channel.go

@@ -0,0 +1,9 @@
+package sink
+
+//go:generate enumer -type Channel -trimprefix Channel -transform snake
+type Channel int
+
+const (
+	ChannelMilestone Channel = iota
+	ChannelQuota
+)

internal/integration/sink/channel_enumer.go

@@ -0,0 +1,78 @@
+// Code generated by "enumer -type Channel -trimprefix Channel -transform snake"; DO NOT EDIT.
+
+package sink
+
+import (
+	"fmt"
+	"strings"
+)
+
+const _ChannelName = "milestonequota"
+
+var _ChannelIndex = [...]uint8{0, 9, 14}
+
+const _ChannelLowerName = "milestonequota"
+
+func (i Channel) String() string {
+	if i < 0 || i >= Channel(len(_ChannelIndex)-1) {
+		return fmt.Sprintf("Channel(%d)", i)
+	}
+	return _ChannelName[_ChannelIndex[i]:_ChannelIndex[i+1]]
+}
+
+// An "invalid array index" compiler error signifies that the constant values have changed.
+// Re-run the stringer command to generate them again.
+func _ChannelNoOp() {
+	var x [1]struct{}
+	_ = x[ChannelMilestone-(0)]
+	_ = x[ChannelQuota-(1)]
+}
+
+var _ChannelValues = []Channel{ChannelMilestone, ChannelQuota}
+
+var _ChannelNameToValueMap = map[string]Channel{
+	_ChannelName[0:9]:       ChannelMilestone,
+	_ChannelLowerName[0:9]:  ChannelMilestone,
+	_ChannelName[9:14]:      ChannelQuota,
+	_ChannelLowerName[9:14]: ChannelQuota,
+}
+
+var _ChannelNames = []string{
+	_ChannelName[0:9],
+	_ChannelName[9:14],
+}
+
+// ChannelString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func ChannelString(s string) (Channel, error) {
+	if val, ok := _ChannelNameToValueMap[s]; ok {
+		return val, nil
+	}
+
+	if val, ok := _ChannelNameToValueMap[strings.ToLower(s)]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to Channel values", s)
+}
+
+// ChannelValues returns all values of the enum
+func ChannelValues() []Channel {
+	return _ChannelValues
+}
+
+// ChannelStrings returns a slice of all String values of the enum
+func ChannelStrings() []string {
+	strs := make([]string, len(_ChannelNames))
+	copy(strs, _ChannelNames)
+	return strs
+}
+
+// IsAChannel returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i Channel) IsAChannel() bool {
+	for _, v := range _ChannelValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

internal/integration/sink/server.go

@@ -0,0 +1,167 @@
+//go:build integration
+
+package sink
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"net/url"
+	"path"
+	"sync"
+	"sync/atomic"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/gorilla/websocket"
+	"github.com/sirupsen/logrus"
+	"github.com/zitadel/logging"
+)
+
+const (
+	port       = "8081"
+	listenAddr = "127.0.0.1:" + port
+	host       = "localhost:" + port
+)
+
+// CallURL returns the full URL to the handler of a [Channel].
+func CallURL(ch Channel) string {
+	u := url.URL{
+		Scheme: "http",
+		Host:   host,
+		Path:   rootPath(ch),
+	}
+	return u.String()
+}
+
+// StartServer starts a simple HTTP server on localhost:8081
+// ZITADEL can use the server to send HTTP requests which can be
+// used to validate tests through [Subscribe]rs.
+// For each [Channel] a route is registered on http://localhost:8081/<channel_name>.
+// The route must be used to send the HTTP request to be validated.
+// [CallURL] can be used to obtain the full URL for a given Channel.
+//
+// This function is only active when the `integration` build tag is enabled
+func StartServer() (close func()) {
+	router := chi.NewRouter()
+	for _, ch := range ChannelValues() {
+		fwd := &forwarder{
+			channelID:   ch,
+			subscribers: make(map[int64]chan<- *Request),
+		}
+		router.HandleFunc(rootPath(ch), fwd.receiveHandler)
+		router.HandleFunc(subscribePath(ch), fwd.subscriptionHandler)
+	}
+	s := &http.Server{
+		Addr:    listenAddr,
+		Handler: router,
+	}
+
+	logging.WithFields("listen_addr", listenAddr).Warn("!!!! A sink server is started which may expose sensitive data on a public endpoint. Make sure the `integration` build tag is disabled for production builds. !!!!")
+	go func() {
+		err := s.ListenAndServe()
+		if !errors.Is(err, http.ErrServerClosed) {
+			logging.WithError(err).Fatal("sink server")
+		}
+	}()
+	return func() {
+		logging.OnError(s.Close()).Error("sink server")
+	}
+}
+
+func rootPath(c Channel) string {
+	return path.Join("/", c.String())
+}
+
+func subscribePath(c Channel) string {
+	return path.Join("/", c.String(), "subscribe")
+}
+
+// forwarder handles incoming HTTP requests from ZITADEL and
+// forwards them to all subscribed web sockets.
+type forwarder struct {
+	channelID   Channel
+	id          atomic.Int64
+	mtx         sync.RWMutex
+	subscribers map[int64]chan<- *Request
+	upgrader    websocket.Upgrader
+}
+
+// receiveHandler receives a simple HTTP for a single [Channel]
+// and forwards them on all active subscribers of that Channel.
+func (c *forwarder) receiveHandler(w http.ResponseWriter, r *http.Request) {
+	req := &Request{
+		Header: r.Header.Clone(),
+	}
+	var err error
+	req.Body, err = io.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+	}
+
+	c.mtx.RLock()
+	for _, reqChan := range c.subscribers {
+		reqChan <- req
+	}
+	c.mtx.RUnlock()
+	w.WriteHeader(http.StatusOK)
+}
+
+// subscriptionHandler upgrades HTTP request to a websocket connection for subscribers.
+// All received HTTP requests on a subscriber's channel are send on the websocket to the client.
+func (c *forwarder) subscriptionHandler(w http.ResponseWriter, r *http.Request) {
+	ws, err := c.upgrader.Upgrade(w, r, nil)
+	logging.OnError(err).Error("websocket upgrade")
+	if err != nil {
+		return
+	}
+	done := readLoop(ws)
+
+	id := c.id.Add(1)
+	reqChannel := make(chan *Request, 100)
+
+	c.mtx.Lock()
+	c.subscribers[id] = reqChannel
+	c.mtx.Unlock()
+
+	logging.WithFields("id", id, "channel", c.channelID).Info("websocket opened")
+
+	defer func() {
+		c.mtx.Lock()
+		delete(c.subscribers, id)
+		c.mtx.Unlock()
+
+		ws.Close()
+		close(reqChannel)
+	}()
+
+	for {
+		select {
+		case err := <-done:
+			logging.WithError(err).WithFields(logrus.Fields{"id": id, "channel": c.channelID}).Info("websocket closed")
+			return
+		case req := <-reqChannel:
+			if err := ws.WriteJSON(req); err != nil {
+				logging.WithError(err).WithFields(logrus.Fields{"id": id, "channel": c.channelID}).Error("websocket write json")
+				return
+			}
+		}
+	}
+}
+
+// readLoop makes sure we can receive close messages
+func readLoop(ws *websocket.Conn) (done chan error) {
+	done = make(chan error, 1)
+
+	go func(done chan<- error) {
+		for {
+			_, _, err := ws.NextReader()
+			if err != nil {
+				done <- err
+				break
+			}
+		}
+		close(done)
+	}(done)
+
+	return done
+}

internal/integration/sink/sink.go

@@ -0,0 +1,4 @@
+// Package sink provides a simple HTTP server where Zitadel can send HTTP based messages,
+// which are then possible to be observed using observers on websockets.
+// The contents of this package become available when the `integration` build tag is enabled.
+package sink

internal/integration/sink/stub.go

@@ -0,0 +1,9 @@
+//go:build !integration
+
+package sink
+
+// StartServer and its returned close function are a no-op
+// when the `integration` build tag is disabled.
+func StartServer() (close func()) {
+	return func() {}
+}

internal/integration/sink/subscription.go

@@ -0,0 +1,90 @@
+//go:build integration
+
+package sink
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"sync/atomic"
+
+	"github.com/gorilla/websocket"
+	"github.com/zitadel/logging"
+)
+
+// Request is a message forwarded from the handler to [Subscription]s.
+type Request struct {
+	Header http.Header
+	Body   json.RawMessage
+}
+
+// Subscription is a websocket client to which [Request]s are forwarded by the server.
+type Subscription struct {
+	conn       *websocket.Conn
+	closed     atomic.Bool
+	reqChannel chan *Request
+}
+
+// Subscribe to a channel.
+// The subscription forwards all requests it received on the channel's
+// handler, after Subscribe has returned.
+// Multiple subscription may be active on a single channel.
+// Each request is always forwarded to each Subscription.
+// Close must be called to cleanup up the Subscription's channel and go routine.
+func Subscribe(ctx context.Context, ch Channel) *Subscription {
+	u := url.URL{
+		Scheme: "ws",
+		Host:   listenAddr,
+		Path:   subscribePath(ch),
+	}
+	conn, resp, err := websocket.DefaultDialer.DialContext(ctx, u.String(), nil)
+	if err != nil {
+		if resp != nil {
+			defer resp.Body.Close()
+			body, _ := io.ReadAll(resp.Body)
+			err = fmt.Errorf("subscribe: %w, status: %s, body: %s", err, resp.Status, body)
+		}
+		panic(err)
+	}
+
+	sub := &Subscription{
+		conn:       conn,
+		reqChannel: make(chan *Request, 10),
+	}
+	go sub.readToChan()
+	return sub
+}
+
+func (s *Subscription) readToChan() {
+	for {
+		if s.closed.Load() {
+			break
+		}
+		req := new(Request)
+		if err := s.conn.ReadJSON(req); err != nil {
+			opErr := new(net.OpError)
+			if errors.As(err, &opErr) {
+				break
+			}
+			logging.WithError(err).Error("subscription read")
+			break
+		}
+		s.reqChannel <- req
+	}
+	close(s.reqChannel)
+}
+
+// Recv returns the channel over which [Request]s are send.
+func (s *Subscription) Recv() <-chan *Request {
+	return s.reqChannel
+}
+
+func (s *Subscription) Close() error {
+	s.closed.Store(true)
+	return s.conn.Close()
+}