Proxy: Support proxy server for SRS. v7.0.16 (#4158)

Please note that the proxy server is a new architecture or the next
version of the Origin Cluster, which allows the publication of multiple
streams. The SRS origin cluster consists of a group of origin servers
designed to handle a large number of streams.

```text
                         +-----------------------+
                     +---+ SRS Proxy(Deployment) +------+---------------------+
+-----------------+  |   +-----------+-----------+      +                     +
| LB(K8s Service) +--+               +(Redis/MESH)      + SRS Origin Servers  +
+-----------------+  |   +-----------+-----------+      +    (Deployment)     +
                     +---+ SRS Proxy(Deployment) +------+---------------------+
                         +-----------------------+
```

The new origin cluster is designed as a collection of proxy servers. For
more information, see [Discussion
#3634](#3634). If you prefer to
use the old origin cluster, please switch to a version before SRS 6.0.

A proxy server can be used for a set of origin servers, which are
isolated and dedicated origin servers. The main improvement in the new
architecture is to store the state for origin servers in the proxy
server, rather than using MESH to communicate between origin servers.
With a proxy server, you can deploy origin servers as stateless servers,
such as in a Kubernetes (K8s) deployment.

Now that the proxy server is a stateful server, it uses Redis to store
the states. For faster development, we use Go to develop the proxy
server, instead of C/C++. Therefore, the proxy server itself is also
stateless, with all states stored in the Redis server or cluster. This
makes the new origin cluster architecture very powerful and robust.

The proxy server is also an architecture designed to solve multiple
process bottlenecks. You can run hundreds of SRS origin servers with one
proxy server on the same machine. This solution can utilize multi-core
machines, such as servers with 128 CPUs. Thus, we can keep SRS
single-threaded and very simple. See
#3665 (comment)
for details.

```text
                                       +--------------------+
                               +-------+ SRS Origin Server  +
                               +       +--------------------+
                               +
+-----------------------+      +       +--------------------+
+ SRS Proxy(Deployment) +------+-------+ SRS Origin Server  +
+-----------------------+      +       +--------------------+
                               +
                               +       +--------------------+
                               +-------+ SRS Origin Server  +
                                       +--------------------+
```

Keep in mind that the proxy server for the Origin Cluster is designed to
handle many streams. To address the issue of many viewers, we will
enhance the Edge Cluster to support more protocols.

```text
+------------------+                                               +--------------------+
+ SRS Edge Server  +--+                                    +-------+ SRS Origin Server  +
+------------------+  +                                    +       +--------------------+
                      +                                    +
+------------------+  +     +-----------------------+      +       +--------------------+
+ SRS Edge Server  +--+-----+ SRS Proxy(Deployment) +------+-------+ SRS Origin Server  +
+------------------+  +     +-----------------------+      +       +--------------------+
                      +                                    +
+------------------+  +                                    +       +--------------------+
+ SRS Edge Server  +--+                                    +-------+ SRS Origin Server  +
+------------------+                                               +--------------------+
```

With the new Origin Cluster and Edge Cluster, you have a media system
capable of supporting a large number of streams and viewers. For
example, you can publish 10,000 streams, each with 100,000 viewers.

---------

Co-authored-by: Jacob Su <[email protected]>

proxy/.gitignore

@@ -0,0 +1,4 @@
+.idea
+srs-proxy
+.env
+.go-formarted

proxy/Makefile

@@ -0,0 +1,23 @@
+.PHONY: all build test fmt clean run
+
+all: build
+
+build: fmt ./srs-proxy
+
+./srs-proxy: *.go
+	go build -o srs-proxy .
+
+test:
+	go test ./...
+
+fmt: ./.go-formarted
+
+./.go-formarted: *.go
+	touch .go-formarted
+	go fmt ./...
+
+clean:
+	rm -f srs-proxy .go-formarted
+
+run: fmt
+	go run .

proxy/api.go

@@ -0,0 +1,272 @@
+// Copyright (c) 2024 Winlin
+//
+// SPDX-License-Identifier: MIT
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"srs-proxy/errors"
+	"srs-proxy/logger"
+)
+
+// srsHTTPAPIServer is the proxy for SRS HTTP API, to proxy the WebRTC HTTP API like WHIP and WHEP,
+// to proxy other HTTP API of SRS like the streams and clients, etc.
+type srsHTTPAPIServer struct {
+	// The underlayer HTTP server.
+	server *http.Server
+	// The WebRTC server.
+	rtc *srsWebRTCServer
+	// The gracefully quit timeout, wait server to quit.
+	gracefulQuitTimeout time.Duration
+	// The wait group for all goroutines.
+	wg sync.WaitGroup
+}
+
+func NewSRSHTTPAPIServer(opts ...func(*srsHTTPAPIServer)) *srsHTTPAPIServer {
+	v := &srsHTTPAPIServer{}
+	for _, opt := range opts {
+		opt(v)
+	}
+	return v
+}
+
+func (v *srsHTTPAPIServer) Close() error {
+	ctx, cancel := context.WithTimeout(context.Background(), v.gracefulQuitTimeout)
+	defer cancel()
+	v.server.Shutdown(ctx)
+
+	v.wg.Wait()
+	return nil
+}
+
+func (v *srsHTTPAPIServer) Run(ctx context.Context) error {
+	// Parse address to listen.
+	addr := envHttpAPI()
+	if !strings.Contains(addr, ":") {
+		addr = ":" + addr
+	}
+
+	// Create server and handler.
+	mux := http.NewServeMux()
+	v.server = &http.Server{Addr: addr, Handler: mux}
+	logger.Df(ctx, "HTTP API server listen at %v", addr)
+
+	// Shutdown the server gracefully when quiting.
+	go func() {
+		ctxParent := ctx
+		<-ctxParent.Done()
+
+		ctx, cancel := context.WithTimeout(context.Background(), v.gracefulQuitTimeout)
+		defer cancel()
+
+		v.server.Shutdown(ctx)
+	}()
+
+	// The basic version handler, also can be used as health check API.
+	logger.Df(ctx, "Handle /api/v1/versions by %v", addr)
+	mux.HandleFunc("/api/v1/versions", func(w http.ResponseWriter, r *http.Request) {
+		apiResponse(ctx, w, r, map[string]string{
+			"signature": Signature(),
+			"version":   Version(),
+		})
+	})
+
+	// The WebRTC WHIP API handler.
+	logger.Df(ctx, "Handle /rtc/v1/whip/ by %v", addr)
+	mux.HandleFunc("/rtc/v1/whip/", func(w http.ResponseWriter, r *http.Request) {
+		if err := v.rtc.HandleApiForWHIP(ctx, w, r); err != nil {
+			apiError(ctx, w, r, err)
+		}
+	})
+
+	// The WebRTC WHEP API handler.
+	logger.Df(ctx, "Handle /rtc/v1/whep/ by %v", addr)
+	mux.HandleFunc("/rtc/v1/whep/", func(w http.ResponseWriter, r *http.Request) {
+		if err := v.rtc.HandleApiForWHEP(ctx, w, r); err != nil {
+			apiError(ctx, w, r, err)
+		}
+	})
+
+	// Run HTTP API server.
+	v.wg.Add(1)
+	go func() {
+		defer v.wg.Done()
+
+		err := v.server.ListenAndServe()
+		if err != nil {
+			if ctx.Err() != context.Canceled {
+				// TODO: If HTTP API server closed unexpectedly, we should notice the main loop to quit.
+				logger.Wf(ctx, "HTTP API accept err %+v", err)
+			} else {
+				logger.Df(ctx, "HTTP API server done")
+			}
+		}
+	}()
+
+	return nil
+}
+
+// systemAPI is the system HTTP API of the proxy server, for SRS media server to register the service
+// to proxy server. It also provides some other system APIs like the status of proxy server, like exporter
+// for Prometheus metrics.
+type systemAPI struct {
+	// The underlayer HTTP server.
+	server *http.Server
+	// The gracefully quit timeout, wait server to quit.
+	gracefulQuitTimeout time.Duration
+	// The wait group for all goroutines.
+	wg sync.WaitGroup
+}
+
+func NewSystemAPI(opts ...func(*systemAPI)) *systemAPI {
+	v := &systemAPI{}
+	for _, opt := range opts {
+		opt(v)
+	}
+	return v
+}
+
+func (v *systemAPI) Close() error {
+	ctx, cancel := context.WithTimeout(context.Background(), v.gracefulQuitTimeout)
+	defer cancel()
+	v.server.Shutdown(ctx)
+
+	v.wg.Wait()
+	return nil
+}
+
+func (v *systemAPI) Run(ctx context.Context) error {
+	// Parse address to listen.
+	addr := envSystemAPI()
+	if !strings.Contains(addr, ":") {
+		addr = ":" + addr
+	}
+
+	// Create server and handler.
+	mux := http.NewServeMux()
+	v.server = &http.Server{Addr: addr, Handler: mux}
+	logger.Df(ctx, "System API server listen at %v", addr)
+
+	// Shutdown the server gracefully when quiting.
+	go func() {
+		ctxParent := ctx
+		<-ctxParent.Done()
+
+		ctx, cancel := context.WithTimeout(context.Background(), v.gracefulQuitTimeout)
+		defer cancel()
+
+		v.server.Shutdown(ctx)
+	}()
+
+	// The basic version handler, also can be used as health check API.
+	logger.Df(ctx, "Handle /api/v1/versions by %v", addr)
+	mux.HandleFunc("/api/v1/versions", func(w http.ResponseWriter, r *http.Request) {
+		apiResponse(ctx, w, r, map[string]string{
+			"signature": Signature(),
+			"version":   Version(),
+		})
+	})
+
+	// The register service for SRS media servers.
+	logger.Df(ctx, "Handle /api/v1/srs/register by %v", addr)
+	mux.HandleFunc("/api/v1/srs/register", func(w http.ResponseWriter, r *http.Request) {
+		if err := func() error {
+			var deviceID, ip, serverID, serviceID, pid string
+			var rtmp, stream, api, srt, rtc []string
+			if err := ParseBody(r.Body, &struct {
+				// The IP of SRS, mandatory.
+				IP *string `json:"ip"`
+				// The server id of SRS, store in file, may not change, mandatory.
+				ServerID *string `json:"server"`
+				// The service id of SRS, always change when restarted, mandatory.
+				ServiceID *string `json:"service"`
+				// The process id of SRS, always change when restarted, mandatory.
+				PID *string `json:"pid"`
+				// The RTMP listen endpoints, mandatory.
+				RTMP *[]string `json:"rtmp"`
+				// The HTTP Stream listen endpoints, optional.
+				HTTP *[]string `json:"http"`
+				// The API listen endpoints, optional.
+				API *[]string `json:"api"`
+				// The SRT listen endpoints, optional.
+				SRT *[]string `json:"srt"`
+				// The RTC listen endpoints, optional.
+				RTC *[]string `json:"rtc"`
+				// The device id of SRS, optional.
+				DeviceID *string `json:"device_id"`
+			}{
+				IP: &ip, DeviceID: &deviceID,
+				ServerID: &serverID, ServiceID: &serviceID, PID: &pid,
+				RTMP: &rtmp, HTTP: &stream, API: &api, SRT: &srt, RTC: &rtc,
+			}); err != nil {
+				return errors.Wrapf(err, "parse body")
+			}
+
+			if ip == "" {
+				return errors.Errorf("empty ip")
+			}
+			if serverID == "" {
+				return errors.Errorf("empty server")
+			}
+			if serviceID == "" {
+				return errors.Errorf("empty service")
+			}
+			if pid == "" {
+				return errors.Errorf("empty pid")
+			}
+			if len(rtmp) == 0 {
+				return errors.Errorf("empty rtmp")
+			}
+
+			server := NewSRSServer(func(srs *SRSServer) {
+				srs.IP, srs.DeviceID = ip, deviceID
+				srs.ServerID, srs.ServiceID, srs.PID = serverID, serviceID, pid
+				srs.RTMP, srs.HTTP, srs.API = rtmp, stream, api
+				srs.SRT, srs.RTC = srt, rtc
+				srs.UpdatedAt = time.Now()
+			})
+			if err := srsLoadBalancer.Update(ctx, server); err != nil {
+				return errors.Wrapf(err, "update SRS server %+v", server)
+			}
+
+			logger.Df(ctx, "Register SRS media server, %+v", server)
+			return nil
+		}(); err != nil {
+			apiError(ctx, w, r, err)
+		}
+
+		type Response struct {
+			Code int    `json:"code"`
+			PID  string `json:"pid"`
+		}
+
+		apiResponse(ctx, w, r, &Response{
+			Code: 0, PID: fmt.Sprintf("%v", os.Getpid()),
+		})
+	})
+
+	// Run System API server.
+	v.wg.Add(1)
+	go func() {
+		defer v.wg.Done()
+
+		err := v.server.ListenAndServe()
+		if err != nil {
+			if ctx.Err() != context.Canceled {
+				// TODO: If System API server closed unexpectedly, we should notice the main loop to quit.
+				logger.Wf(ctx, "System API accept err %+v", err)
+			} else {
+				logger.Df(ctx, "System API server done")
+			}
+		}
+	}()
+
+	return nil
+}

proxy/debug.go

@@ -0,0 +1,20 @@
+// Copyright (c) 2024 Winlin
+//
+// SPDX-License-Identifier: MIT
+package main
+
+import (
+	"context"
+	"net/http"
+
+	"srs-proxy/logger"
+)
+
+func handleGoPprof(ctx context.Context) {
+	if addr := envGoPprof(); addr != "" {
+		go func() {
+			logger.Df(ctx, "Start Go pprof at %v", addr)
+			http.ListenAndServe(addr, nil)
+		}()
+	}
+}