mirror of
https://github.com/matrix-org/dendrite.git
synced 2024-11-27 00:31:55 -06:00
d8b19c857f
Beforehand we disabled HTTP keepalives to prevent ambient system resources from being used by excess idle connections. Now that we've fixed some bugs in the federation API and device list updater, this situation is now much better and we don't open so many remote connections anyway. Keepalives allow us to not have to handshake TLS so often (which is quite expensive) and reusing an idle connection is much faster than having to open a new one. This can help with response times when talking to remote federated servers. This PR also adds a new option to disable keepalives if needed: ``` # Disable HTTP keepalives, which also prevents connection reuse. Dendrite will typically # keep HTTP connections open to remote hosts for 5 minutes as they can be reused much # more quickly than opening new connections each time. Disabling keepalives will close # HTTP connections immediately after a successful request but may result in more CPU and # memory being used on TLS handshakes for each new connection instead. disable_http_keepalives: false ```
582 lines
22 KiB
Go
582 lines
22 KiB
Go
// Copyright 2020 The Matrix.org Foundation C.I.C.
|
||
//
|
||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||
// you may not use this file except in compliance with the License.
|
||
// You may obtain a copy of the License at
|
||
//
|
||
// http://www.apache.org/licenses/LICENSE-2.0
|
||
//
|
||
// Unless required by applicable law or agreed to in writing, software
|
||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
// See the License for the specific language governing permissions and
|
||
// limitations under the License.
|
||
|
||
package base
|
||
|
||
import (
|
||
"context"
|
||
"crypto/tls"
|
||
"database/sql"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
_ "net/http/pprof"
|
||
"os"
|
||
"os/signal"
|
||
"sync"
|
||
"syscall"
|
||
"time"
|
||
|
||
"github.com/getsentry/sentry-go"
|
||
sentryhttp "github.com/getsentry/sentry-go/http"
|
||
"github.com/matrix-org/gomatrixserverlib"
|
||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||
"go.uber.org/atomic"
|
||
"golang.org/x/net/http2"
|
||
"golang.org/x/net/http2/h2c"
|
||
|
||
"github.com/matrix-org/dendrite/internal/caching"
|
||
"github.com/matrix-org/dendrite/internal/fulltext"
|
||
"github.com/matrix-org/dendrite/internal/httputil"
|
||
"github.com/matrix-org/dendrite/internal/pushgateway"
|
||
"github.com/matrix-org/dendrite/internal/sqlutil"
|
||
|
||
"github.com/matrix-org/dendrite/internal"
|
||
"github.com/matrix-org/dendrite/setup/jetstream"
|
||
"github.com/matrix-org/dendrite/setup/process"
|
||
|
||
"github.com/gorilla/mux"
|
||
"github.com/kardianos/minwinsvc"
|
||
|
||
"github.com/sirupsen/logrus"
|
||
|
||
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
|
||
asinthttp "github.com/matrix-org/dendrite/appservice/inthttp"
|
||
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
|
||
federationIntHTTP "github.com/matrix-org/dendrite/federationapi/inthttp"
|
||
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
|
||
keyinthttp "github.com/matrix-org/dendrite/keyserver/inthttp"
|
||
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
|
||
rsinthttp "github.com/matrix-org/dendrite/roomserver/inthttp"
|
||
"github.com/matrix-org/dendrite/setup/config"
|
||
userapi "github.com/matrix-org/dendrite/userapi/api"
|
||
userapiinthttp "github.com/matrix-org/dendrite/userapi/inthttp"
|
||
)
|
||
|
||
// BaseDendrite is a base for creating new instances of dendrite. It parses
|
||
// command line flags and config, and exposes methods for creating various
|
||
// resources. All errors are handled by logging then exiting, so all methods
|
||
// should only be used during start up.
|
||
// Must be closed when shutting down.
|
||
type BaseDendrite struct {
|
||
*process.ProcessContext
|
||
componentName string
|
||
tracerCloser io.Closer
|
||
PublicClientAPIMux *mux.Router
|
||
PublicFederationAPIMux *mux.Router
|
||
PublicKeyAPIMux *mux.Router
|
||
PublicMediaAPIMux *mux.Router
|
||
PublicWellKnownAPIMux *mux.Router
|
||
InternalAPIMux *mux.Router
|
||
DendriteAdminMux *mux.Router
|
||
SynapseAdminMux *mux.Router
|
||
NATS *jetstream.NATSInstance
|
||
UseHTTPAPIs bool
|
||
apiHttpClient *http.Client
|
||
Cfg *config.Dendrite
|
||
Caches *caching.Caches
|
||
DNSCache *gomatrixserverlib.DNSCache
|
||
Database *sql.DB
|
||
DatabaseWriter sqlutil.Writer
|
||
EnableMetrics bool
|
||
Fulltext *fulltext.Search
|
||
startupLock sync.Mutex
|
||
}
|
||
|
||
const NoListener = ""
|
||
|
||
const HTTPServerTimeout = time.Minute * 5
|
||
const HTTPClientTimeout = time.Second * 30
|
||
|
||
type BaseDendriteOptions int
|
||
|
||
const (
|
||
DisableMetrics BaseDendriteOptions = iota
|
||
UseHTTPAPIs
|
||
PolylithMode
|
||
)
|
||
|
||
// NewBaseDendrite creates a new instance to be used by a component.
|
||
// The componentName is used for logging purposes, and should be a friendly name
|
||
// of the compontent running, e.g. "SyncAPI"
|
||
func NewBaseDendrite(cfg *config.Dendrite, componentName string, options ...BaseDendriteOptions) *BaseDendrite {
|
||
platformSanityChecks()
|
||
useHTTPAPIs := false
|
||
enableMetrics := true
|
||
isMonolith := true
|
||
for _, opt := range options {
|
||
switch opt {
|
||
case DisableMetrics:
|
||
enableMetrics = false
|
||
case UseHTTPAPIs:
|
||
useHTTPAPIs = true
|
||
case PolylithMode:
|
||
isMonolith = false
|
||
useHTTPAPIs = true
|
||
}
|
||
}
|
||
|
||
configErrors := &config.ConfigErrors{}
|
||
cfg.Verify(configErrors, isMonolith)
|
||
if len(*configErrors) > 0 {
|
||
for _, err := range *configErrors {
|
||
logrus.Errorf("Configuration error: %s", err)
|
||
}
|
||
logrus.Fatalf("Failed to start due to configuration errors")
|
||
}
|
||
|
||
internal.SetupStdLogging()
|
||
internal.SetupHookLogging(cfg.Logging, componentName)
|
||
internal.SetupPprof()
|
||
|
||
logrus.Infof("Dendrite version %s", internal.VersionString())
|
||
|
||
if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
|
||
logrus.Warn("Open registration is enabled")
|
||
}
|
||
|
||
closer, err := cfg.SetupTracing("Dendrite" + componentName)
|
||
if err != nil {
|
||
logrus.WithError(err).Panicf("failed to start opentracing")
|
||
}
|
||
|
||
var fts *fulltext.Search
|
||
isSyncOrMonolith := componentName == "syncapi" || isMonolith
|
||
if cfg.SyncAPI.Fulltext.Enabled && isSyncOrMonolith {
|
||
fts, err = fulltext.New(cfg.SyncAPI.Fulltext)
|
||
if err != nil {
|
||
logrus.WithError(err).Panicf("failed to create full text")
|
||
}
|
||
}
|
||
|
||
if cfg.Global.Sentry.Enabled {
|
||
logrus.Info("Setting up Sentry for debugging...")
|
||
err = sentry.Init(sentry.ClientOptions{
|
||
Dsn: cfg.Global.Sentry.DSN,
|
||
Environment: cfg.Global.Sentry.Environment,
|
||
Debug: true,
|
||
ServerName: string(cfg.Global.ServerName),
|
||
Release: "dendrite@" + internal.VersionString(),
|
||
AttachStacktrace: true,
|
||
})
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("failed to start Sentry")
|
||
}
|
||
}
|
||
|
||
var dnsCache *gomatrixserverlib.DNSCache
|
||
if cfg.Global.DNSCache.Enabled {
|
||
dnsCache = gomatrixserverlib.NewDNSCache(
|
||
cfg.Global.DNSCache.CacheSize,
|
||
cfg.Global.DNSCache.CacheLifetime,
|
||
)
|
||
logrus.Infof(
|
||
"DNS cache enabled (size %d, lifetime %s)",
|
||
cfg.Global.DNSCache.CacheSize,
|
||
cfg.Global.DNSCache.CacheLifetime,
|
||
)
|
||
}
|
||
|
||
apiClient := http.Client{
|
||
Timeout: time.Minute * 10,
|
||
Transport: &http2.Transport{
|
||
AllowHTTP: true,
|
||
DialTLS: func(network, addr string, _ *tls.Config) (net.Conn, error) {
|
||
// Ordinarily HTTP/2 would expect TLS, but the remote listener is
|
||
// H2C-enabled (HTTP/2 without encryption). Overriding the DialTLS
|
||
// function with a plain Dial allows us to trick the HTTP client
|
||
// into establishing a HTTP/2 connection without TLS.
|
||
// TODO: Eventually we will want to look at authenticating and
|
||
// encrypting these internal HTTP APIs, at which point we will have
|
||
// to reconsider H2C and change all this anyway.
|
||
return net.Dial(network, addr)
|
||
},
|
||
},
|
||
}
|
||
|
||
// If we're in monolith mode, we'll set up a global pool of database
|
||
// connections. A component is welcome to use this pool if they don't
|
||
// have a separate database config of their own.
|
||
var db *sql.DB
|
||
var writer sqlutil.Writer
|
||
if cfg.Global.DatabaseOptions.ConnectionString != "" {
|
||
if !isMonolith {
|
||
logrus.Panic("Using a global database connection pool is not supported in polylith deployments")
|
||
}
|
||
if cfg.Global.DatabaseOptions.ConnectionString.IsSQLite() {
|
||
logrus.Panic("Using a global database connection pool is not supported with SQLite databases")
|
||
}
|
||
writer = sqlutil.NewDummyWriter()
|
||
if db, err = sqlutil.Open(&cfg.Global.DatabaseOptions, writer); err != nil {
|
||
logrus.WithError(err).Panic("Failed to set up global database connections")
|
||
}
|
||
logrus.Debug("Using global database connection pool")
|
||
}
|
||
|
||
// Ideally we would only use SkipClean on routes which we know can allow '/' but due to
|
||
// https://github.com/gorilla/mux/issues/460 we have to attach this at the top router.
|
||
// When used in conjunction with UseEncodedPath() we get the behaviour we want when parsing
|
||
// path parameters:
|
||
// /foo/bar%2Fbaz == [foo, bar%2Fbaz] (from UseEncodedPath)
|
||
// /foo/bar%2F%2Fbaz == [foo, bar%2F%2Fbaz] (from SkipClean)
|
||
// In particular, rooms v3 event IDs are not urlsafe and can include '/' and because they
|
||
// are randomly generated it results in flakey tests.
|
||
// We need to be careful with media APIs if they read from a filesystem to make sure they
|
||
// are not inadvertently reading paths without cleaning, else this could introduce a
|
||
// directory traversal attack e.g /../../../etc/passwd
|
||
|
||
return &BaseDendrite{
|
||
ProcessContext: process.NewProcessContext(),
|
||
componentName: componentName,
|
||
UseHTTPAPIs: useHTTPAPIs,
|
||
tracerCloser: closer,
|
||
Cfg: cfg,
|
||
Caches: caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, enableMetrics),
|
||
DNSCache: dnsCache,
|
||
PublicClientAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicClientPathPrefix).Subrouter().UseEncodedPath(),
|
||
PublicFederationAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicFederationPathPrefix).Subrouter().UseEncodedPath(),
|
||
PublicKeyAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicKeyPathPrefix).Subrouter().UseEncodedPath(),
|
||
PublicMediaAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicMediaPathPrefix).Subrouter().UseEncodedPath(),
|
||
PublicWellKnownAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.PublicWellKnownPrefix).Subrouter().UseEncodedPath(),
|
||
InternalAPIMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.InternalPathPrefix).Subrouter().UseEncodedPath(),
|
||
DendriteAdminMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.DendriteAdminPathPrefix).Subrouter().UseEncodedPath(),
|
||
SynapseAdminMux: mux.NewRouter().SkipClean(true).PathPrefix(httputil.SynapseAdminPathPrefix).Subrouter().UseEncodedPath(),
|
||
NATS: &jetstream.NATSInstance{},
|
||
apiHttpClient: &apiClient,
|
||
Database: db, // set if monolith with global connection pool only
|
||
DatabaseWriter: writer, // set if monolith with global connection pool only
|
||
EnableMetrics: enableMetrics,
|
||
Fulltext: fts,
|
||
}
|
||
}
|
||
|
||
// Close implements io.Closer
|
||
func (b *BaseDendrite) Close() error {
|
||
return b.tracerCloser.Close()
|
||
}
|
||
|
||
// DatabaseConnection assists in setting up a database connection. It accepts
|
||
// the database properties and a new writer for the given component. If we're
|
||
// running in monolith mode with a global connection pool configured then we
|
||
// will return that connection, along with the global writer, effectively
|
||
// ignoring the options provided. Otherwise we'll open a new database connection
|
||
// using the supplied options and writer. Note that it's possible for the pointer
|
||
// receiver to be nil here – that's deliberate as some of the unit tests don't
|
||
// have a BaseDendrite and just want a connection with the supplied config
|
||
// without any pooling stuff.
|
||
func (b *BaseDendrite) DatabaseConnection(dbProperties *config.DatabaseOptions, writer sqlutil.Writer) (*sql.DB, sqlutil.Writer, error) {
|
||
if dbProperties.ConnectionString != "" || b == nil {
|
||
// Open a new database connection using the supplied config.
|
||
db, err := sqlutil.Open(dbProperties, writer)
|
||
return db, writer, err
|
||
}
|
||
if b.Database != nil && b.DatabaseWriter != nil {
|
||
// Ignore the supplied config and return the global pool and
|
||
// writer.
|
||
return b.Database, b.DatabaseWriter, nil
|
||
}
|
||
return nil, nil, fmt.Errorf("no database connections configured")
|
||
}
|
||
|
||
// AppserviceHTTPClient returns the AppServiceInternalAPI for hitting the appservice component over HTTP.
|
||
func (b *BaseDendrite) AppserviceHTTPClient() appserviceAPI.AppServiceInternalAPI {
|
||
a, err := asinthttp.NewAppserviceClient(b.Cfg.AppServiceURL(), b.apiHttpClient)
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("CreateHTTPAppServiceAPIs failed")
|
||
}
|
||
return a
|
||
}
|
||
|
||
// RoomserverHTTPClient returns RoomserverInternalAPI for hitting the roomserver over HTTP.
|
||
func (b *BaseDendrite) RoomserverHTTPClient() roomserverAPI.RoomserverInternalAPI {
|
||
rsAPI, err := rsinthttp.NewRoomserverClient(b.Cfg.RoomServerURL(), b.apiHttpClient, b.Caches)
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("RoomserverHTTPClient failed", b.apiHttpClient)
|
||
}
|
||
return rsAPI
|
||
}
|
||
|
||
// UserAPIClient returns UserInternalAPI for hitting the userapi over HTTP.
|
||
func (b *BaseDendrite) UserAPIClient() userapi.UserInternalAPI {
|
||
userAPI, err := userapiinthttp.NewUserAPIClient(b.Cfg.UserAPIURL(), b.apiHttpClient)
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("UserAPIClient failed", b.apiHttpClient)
|
||
}
|
||
return userAPI
|
||
}
|
||
|
||
// FederationAPIHTTPClient returns FederationInternalAPI for hitting
|
||
// the federation API server over HTTP
|
||
func (b *BaseDendrite) FederationAPIHTTPClient() federationAPI.FederationInternalAPI {
|
||
f, err := federationIntHTTP.NewFederationAPIClient(b.Cfg.FederationAPIURL(), b.apiHttpClient, b.Caches)
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("FederationAPIHTTPClient failed", b.apiHttpClient)
|
||
}
|
||
return f
|
||
}
|
||
|
||
// KeyServerHTTPClient returns KeyInternalAPI for hitting the key server over HTTP
|
||
func (b *BaseDendrite) KeyServerHTTPClient() keyserverAPI.KeyInternalAPI {
|
||
f, err := keyinthttp.NewKeyServerClient(b.Cfg.KeyServerURL(), b.apiHttpClient)
|
||
if err != nil {
|
||
logrus.WithError(err).Panic("KeyServerHTTPClient failed", b.apiHttpClient)
|
||
}
|
||
return f
|
||
}
|
||
|
||
// PushGatewayHTTPClient returns a new client for interacting with (external) Push Gateways.
|
||
func (b *BaseDendrite) PushGatewayHTTPClient() pushgateway.Client {
|
||
return pushgateway.NewHTTPClient(b.Cfg.UserAPI.PushGatewayDisableTLSValidation)
|
||
}
|
||
|
||
// CreateClient creates a new client (normally used for media fetch requests).
|
||
// Should only be called once per component.
|
||
func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
|
||
if b.Cfg.Global.DisableFederation {
|
||
return gomatrixserverlib.NewClient(
|
||
gomatrixserverlib.WithTransport(noOpHTTPTransport),
|
||
)
|
||
}
|
||
opts := []gomatrixserverlib.ClientOption{
|
||
gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
|
||
gomatrixserverlib.WithWellKnownSRVLookups(true),
|
||
}
|
||
if b.Cfg.Global.DNSCache.Enabled {
|
||
opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
|
||
}
|
||
client := gomatrixserverlib.NewClient(opts...)
|
||
client.SetUserAgent(fmt.Sprintf("Dendrite/%s", internal.VersionString()))
|
||
return client
|
||
}
|
||
|
||
// CreateFederationClient creates a new federation client. Should only be called
|
||
// once per component.
|
||
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
|
||
if b.Cfg.Global.DisableFederation {
|
||
return gomatrixserverlib.NewFederationClient(
|
||
b.Cfg.Global.ServerName, b.Cfg.Global.KeyID, b.Cfg.Global.PrivateKey,
|
||
gomatrixserverlib.WithTransport(noOpHTTPTransport),
|
||
)
|
||
}
|
||
opts := []gomatrixserverlib.ClientOption{
|
||
gomatrixserverlib.WithTimeout(time.Minute * 5),
|
||
gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
|
||
gomatrixserverlib.WithKeepAlives(!b.Cfg.FederationAPI.DisableHTTPKeepalives),
|
||
}
|
||
if b.Cfg.Global.DNSCache.Enabled {
|
||
opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
|
||
}
|
||
client := gomatrixserverlib.NewFederationClient(
|
||
b.Cfg.Global.ServerName, b.Cfg.Global.KeyID,
|
||
b.Cfg.Global.PrivateKey, opts...,
|
||
)
|
||
client.SetUserAgent(fmt.Sprintf("Dendrite/%s", internal.VersionString()))
|
||
return client
|
||
}
|
||
|
||
func (b *BaseDendrite) configureHTTPErrors() {
|
||
notAllowedHandler := func(w http.ResponseWriter, r *http.Request) {
|
||
w.WriteHeader(http.StatusMethodNotAllowed)
|
||
_, _ = w.Write([]byte(fmt.Sprintf("405 %s not allowed on this endpoint", r.Method)))
|
||
}
|
||
|
||
notFoundCORSHandler := httputil.WrapHandlerInCORS(http.NotFoundHandler())
|
||
notAllowedCORSHandler := httputil.WrapHandlerInCORS(http.HandlerFunc(notAllowedHandler))
|
||
|
||
for _, router := range []*mux.Router{
|
||
b.PublicClientAPIMux, b.PublicMediaAPIMux,
|
||
b.DendriteAdminMux, b.SynapseAdminMux,
|
||
b.PublicWellKnownAPIMux,
|
||
} {
|
||
router.NotFoundHandler = notFoundCORSHandler
|
||
router.MethodNotAllowedHandler = notAllowedCORSHandler
|
||
}
|
||
}
|
||
|
||
// SetupAndServeHTTP sets up the HTTP server to serve endpoints registered on
|
||
// ApiMux under /api/ and adds a prometheus handler under /metrics.
|
||
func (b *BaseDendrite) SetupAndServeHTTP(
|
||
internalHTTPAddr, externalHTTPAddr config.HTTPAddress,
|
||
certFile, keyFile *string,
|
||
) {
|
||
// Manually unlocked right before actually serving requests,
|
||
// as we don't return from this method (defer doesn't work).
|
||
b.startupLock.Lock()
|
||
internalAddr, _ := internalHTTPAddr.Address()
|
||
externalAddr, _ := externalHTTPAddr.Address()
|
||
|
||
externalRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||
internalRouter := externalRouter
|
||
|
||
externalServ := &http.Server{
|
||
Addr: string(externalAddr),
|
||
WriteTimeout: HTTPServerTimeout,
|
||
Handler: externalRouter,
|
||
BaseContext: func(_ net.Listener) context.Context {
|
||
return b.ProcessContext.Context()
|
||
},
|
||
}
|
||
internalServ := externalServ
|
||
|
||
if internalAddr != NoListener && externalAddr != internalAddr {
|
||
// H2C allows us to accept HTTP/2 connections without TLS
|
||
// encryption. Since we don't currently require any form of
|
||
// authentication or encryption on these internal HTTP APIs,
|
||
// H2C gives us all of the advantages of HTTP/2 (such as
|
||
// stream multiplexing and avoiding head-of-line blocking)
|
||
// without enabling TLS.
|
||
internalH2S := &http2.Server{}
|
||
internalRouter = mux.NewRouter().SkipClean(true).UseEncodedPath()
|
||
internalServ = &http.Server{
|
||
Addr: string(internalAddr),
|
||
Handler: h2c.NewHandler(internalRouter, internalH2S),
|
||
BaseContext: func(_ net.Listener) context.Context {
|
||
return b.ProcessContext.Context()
|
||
},
|
||
}
|
||
}
|
||
|
||
b.configureHTTPErrors()
|
||
|
||
internalRouter.PathPrefix(httputil.InternalPathPrefix).Handler(b.InternalAPIMux)
|
||
if b.Cfg.Global.Metrics.Enabled {
|
||
internalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), b.Cfg.Global.Metrics.BasicAuth))
|
||
}
|
||
|
||
b.DendriteAdminMux.HandleFunc("/monitor/up", func(w http.ResponseWriter, r *http.Request) {
|
||
w.WriteHeader(200)
|
||
})
|
||
b.DendriteAdminMux.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
|
||
if b.ProcessContext.IsDegraded() {
|
||
w.WriteHeader(503)
|
||
return
|
||
}
|
||
w.WriteHeader(200)
|
||
})
|
||
|
||
var clientHandler http.Handler
|
||
clientHandler = b.PublicClientAPIMux
|
||
if b.Cfg.Global.Sentry.Enabled {
|
||
sentryHandler := sentryhttp.New(sentryhttp.Options{
|
||
Repanic: true,
|
||
})
|
||
clientHandler = sentryHandler.Handle(b.PublicClientAPIMux)
|
||
}
|
||
var federationHandler http.Handler
|
||
federationHandler = b.PublicFederationAPIMux
|
||
if b.Cfg.Global.Sentry.Enabled {
|
||
sentryHandler := sentryhttp.New(sentryhttp.Options{
|
||
Repanic: true,
|
||
})
|
||
federationHandler = sentryHandler.Handle(b.PublicFederationAPIMux)
|
||
}
|
||
internalRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(b.DendriteAdminMux)
|
||
externalRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(clientHandler)
|
||
if !b.Cfg.Global.DisableFederation {
|
||
externalRouter.PathPrefix(httputil.PublicKeyPathPrefix).Handler(b.PublicKeyAPIMux)
|
||
externalRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(federationHandler)
|
||
}
|
||
externalRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(b.SynapseAdminMux)
|
||
externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(b.PublicMediaAPIMux)
|
||
externalRouter.PathPrefix(httputil.PublicWellKnownPrefix).Handler(b.PublicWellKnownAPIMux)
|
||
|
||
b.startupLock.Unlock()
|
||
if internalAddr != NoListener && internalAddr != externalAddr {
|
||
go func() {
|
||
var internalShutdown atomic.Bool // RegisterOnShutdown can be called more than once
|
||
logrus.Infof("Starting internal %s listener on %s", b.componentName, internalServ.Addr)
|
||
b.ProcessContext.ComponentStarted()
|
||
internalServ.RegisterOnShutdown(func() {
|
||
if internalShutdown.CompareAndSwap(false, true) {
|
||
b.ProcessContext.ComponentFinished()
|
||
logrus.Infof("Stopped internal HTTP listener")
|
||
}
|
||
})
|
||
if certFile != nil && keyFile != nil {
|
||
if err := internalServ.ListenAndServeTLS(*certFile, *keyFile); err != nil {
|
||
if err != http.ErrServerClosed {
|
||
logrus.WithError(err).Fatal("failed to serve HTTPS")
|
||
}
|
||
}
|
||
} else {
|
||
if err := internalServ.ListenAndServe(); err != nil {
|
||
if err != http.ErrServerClosed {
|
||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||
}
|
||
}
|
||
}
|
||
logrus.Infof("Stopped internal %s listener on %s", b.componentName, internalServ.Addr)
|
||
}()
|
||
}
|
||
|
||
if externalAddr != NoListener {
|
||
go func() {
|
||
var externalShutdown atomic.Bool // RegisterOnShutdown can be called more than once
|
||
logrus.Infof("Starting external %s listener on %s", b.componentName, externalServ.Addr)
|
||
b.ProcessContext.ComponentStarted()
|
||
externalServ.RegisterOnShutdown(func() {
|
||
if externalShutdown.CompareAndSwap(false, true) {
|
||
b.ProcessContext.ComponentFinished()
|
||
logrus.Infof("Stopped external HTTP listener")
|
||
}
|
||
})
|
||
if certFile != nil && keyFile != nil {
|
||
if err := externalServ.ListenAndServeTLS(*certFile, *keyFile); err != nil {
|
||
if err != http.ErrServerClosed {
|
||
logrus.WithError(err).Fatal("failed to serve HTTPS")
|
||
}
|
||
}
|
||
} else {
|
||
if err := externalServ.ListenAndServe(); err != nil {
|
||
if err != http.ErrServerClosed {
|
||
logrus.WithError(err).Fatal("failed to serve HTTP")
|
||
}
|
||
}
|
||
}
|
||
logrus.Infof("Stopped external %s listener on %s", b.componentName, externalServ.Addr)
|
||
}()
|
||
}
|
||
|
||
minwinsvc.SetOnExit(b.ProcessContext.ShutdownDendrite)
|
||
<-b.ProcessContext.WaitForShutdown()
|
||
|
||
logrus.Infof("Stopping HTTP listeners")
|
||
_ = internalServ.Shutdown(context.Background())
|
||
_ = externalServ.Shutdown(context.Background())
|
||
logrus.Infof("Stopped HTTP listeners")
|
||
}
|
||
|
||
func (b *BaseDendrite) WaitForShutdown() {
|
||
sigs := make(chan os.Signal, 1)
|
||
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
|
||
select {
|
||
case <-sigs:
|
||
case <-b.ProcessContext.WaitForShutdown():
|
||
}
|
||
signal.Reset(syscall.SIGINT, syscall.SIGTERM)
|
||
|
||
logrus.Warnf("Shutdown signal received")
|
||
|
||
b.ProcessContext.ShutdownDendrite()
|
||
b.ProcessContext.WaitForComponentsToFinish()
|
||
if b.Cfg.Global.Sentry.Enabled {
|
||
if !sentry.Flush(time.Second * 5) {
|
||
logrus.Warnf("failed to flush all Sentry events!")
|
||
}
|
||
}
|
||
|
||
logrus.Warnf("Dendrite is exiting now")
|
||
}
|