4350203089

Committed 07 Mar 2023 03:08AM UTC coverage: 85.275% (-0.009%) from 85.284%

Build # 4350203089

Build Type

github

Committed by GitHub

Commit Message

Bump golang.org/x/net from 0.0.0-20210614182718-04defd469f4e to 0.7.0

Pull Request Pull Request #227: Bump golang.org/x/net from 0.0.0-20210614182718-04defd469f4e to 0.7.0

Run Details

9492 of 11131 relevant lines covered (85.28%)

6682.46 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.58

/rest/server.go

package rest

import (
        "context"
        "crypto/tls"
        "fmt"
        "net"
        "net/http"
        "net/url"
        "strings"
        "sync"
        "time"

        metricsutil "github.com/go-phorce/dolly/metrics/util"
        "github.com/go-phorce/dolly/netutil"
        "github.com/go-phorce/dolly/rest/ready"
        "github.com/go-phorce/dolly/tasks"
        "github.com/go-phorce/dolly/xhttp"
        "github.com/go-phorce/dolly/xhttp/authz"
        "github.com/go-phorce/dolly/xhttp/header"
        "github.com/go-phorce/dolly/xhttp/httperror"
        "github.com/go-phorce/dolly/xhttp/identity"
        "github.com/go-phorce/dolly/xhttp/marshal"
        "github.com/go-phorce/dolly/xlog"
        "github.com/pkg/errors"
)

var logger = xlog.NewPackageLogger("github.com/go-phorce/dolly", "rest")

// MaxRequestSize specifies max size of regular HTTP Post requests in bytes, 64 Mb
const MaxRequestSize = 64 * 1024 * 1024

const (
        // EvtSourceStatus specifies source for service Status
        EvtSourceStatus = "status"
        // EvtServiceStarted specifies Service Started event
        EvtServiceStarted = "service started"
        // EvtServiceStopped specifies Service Stopped event
        EvtServiceStopped = "service stopped"
)

// ServerEvent specifies server event type
type ServerEvent int

const (
        // ServerStartedEvent is fired on server start
        ServerStartedEvent ServerEvent = iota
        // ServerStoppedEvent is fired after server stopped
        ServerStoppedEvent
        // ServerStoppingEvent is fired before server stopped
        ServerStoppingEvent
)

// ServerEventFunc is a callback to handle server events
type ServerEventFunc func(evt ServerEvent)

// Server is an interface to provide server status
type Server interface {
        http.Handler
        Name() string
        Version() string
        HostName() string
        LocalIP() string
        Port() string
        Protocol() string
        StartedAt() time.Time
        Uptime() time.Duration
        Service(name string) Service
        HTTPConfig() HTTPServerConfig
        TLSConfig() *tls.Config

        // IsReady indicates that all subservices are ready to serve
        IsReady() bool

        // Audit records an auditable event.
        //         source indicates the area that the event was triggered by
        //         eventType indicates the specific event that occured
        //         identity specifies the identity of the user that triggered this event, typically this is <role>/<cn>
        //         contextID specifies the request ContextID that the event was triggered in [this can be used for cross service correlation of logs]
        //         raftIndex indicates the index# of the raft log in RAFT that the event occured in [if applicable]
        //         message contains any additional information about this event that is eventType specific
        Audit(source string,
                eventType string,
                identity string,
                contextID string,
                raftIndex uint64,
                message string)

        AddService(s Service)
        StartHTTP() error
        StopHTTP()

        Scheduler() tasks.Scheduler

        OnEvent(evt ServerEvent, handler ServerEventFunc)
}

// MuxFactory creates http handlers.
type MuxFactory interface {
        NewMux() http.Handler
}

// HTTPServer is responsible for exposing the collection of the services
// as a single HTTP server
type HTTPServer struct {
        Server
        auditor         Auditor
        authz           authz.HTTPAuthz
        identityMapper  identity.ProviderFromRequest
        httpConfig      HTTPServerConfig
        tlsConfig       *tls.Config
        httpServer      *http.Server
        cors            *CORSOptions
        muxFactory      MuxFactory
        hostname        string
        port            string
        ipaddr          string
        version         string
        serving         bool
        startedAt       time.Time
        clientAuth      string
        scheduler       tasks.Scheduler
        services        map[string]Service
        evtHandlers     map[ServerEvent][]ServerEventFunc
        lock            sync.RWMutex
        shutdownTimeout time.Duration
}

// New creates a new instance of the server
func New(
        version string,
        ipaddr string,
        httpConfig HTTPServerConfig,
        tlsConfig *tls.Config,
) (*HTTPServer, error) {
        var err error

        if ipaddr == "" {
                ipaddr, err = netutil.GetLocalIP()
                if err != nil {
                        ipaddr = "127.0.0.1"
                        logger.Errorf("reason=unable_determine_ipaddr, use=%q, err=[%+v]", ipaddr, err)
                }
        }

        s := &HTTPServer{
                services:        map[string]Service{},
                startedAt:       time.Now().UTC(),
                version:         version,
                ipaddr:          ipaddr,
                evtHandlers:     make(map[ServerEvent][]ServerEventFunc),
                clientAuth:      tlsClientAuthToStrMap[tls.NoClientCert],
                httpConfig:      httpConfig,
                hostname:        GetHostName(httpConfig.GetBindAddr()),
                port:            GetPort(httpConfig.GetBindAddr()),
                tlsConfig:       tlsConfig,
                shutdownTimeout: time.Duration(5) * time.Second,
        }
        s.muxFactory = s
        if tlsConfig != nil {
                s.clientAuth = tlsClientAuthToStrMap[tlsConfig.ClientAuth]
        }

        return s, nil
}

// WithAuditor enables to use auditor
func (server *HTTPServer) WithAuditor(auditor Auditor) *HTTPServer {
        server.auditor = auditor
        return server
}

// WithAuthz enables to use Authz
func (server *HTTPServer) WithAuthz(authz authz.HTTPAuthz) *HTTPServer {
        server.authz = authz
        return server
}

// WithIdentityProvider enables to set idenity on each request
func (server *HTTPServer) WithIdentityProvider(provider identity.ProviderFromRequest) *HTTPServer {
        server.identityMapper = provider
        return server
}

// WithScheduler enables to schedule tasks, such as heartbeat, uptime etc
// If a scheduler is not provided, then tasks will not be schedduled.
func (server *HTTPServer) WithScheduler(scheduler tasks.Scheduler) *HTTPServer {
        server.scheduler = scheduler
        return server
}

// WithCORS enables CORS options
func (server *HTTPServer) WithCORS(cors *CORSOptions) *HTTPServer {
        server.cors = cors
        return server
}

// WithShutdownTimeout sets the connection draining timeouts on server shutdown
func (server *HTTPServer) WithShutdownTimeout(timeout time.Duration) *HTTPServer {
        server.shutdownTimeout = timeout
        return server
}

var tlsClientAuthToStrMap = map[tls.ClientAuthType]string{
        tls.NoClientCert:               "NoClientCert",
        tls.RequestClientCert:          "RequestClientCert",
        tls.RequireAnyClientCert:       "RequireAnyClientCert",
        tls.VerifyClientCertIfGiven:    "VerifyClientCertIfGiven",
        tls.RequireAndVerifyClientCert: "RequireAndVerifyClientCert",
}

// AddService provides a service registration for the server
func (server *HTTPServer) AddService(s Service) {
        server.lock.Lock()
        defer server.lock.Unlock()
        server.services[s.Name()] = s
}

// OnEvent accepts a callback to handle server events
func (server *HTTPServer) OnEvent(evt ServerEvent, handler ServerEventFunc) {
        server.lock.Lock()
        defer server.lock.Unlock()

        server.evtHandlers[evt] = append(server.evtHandlers[evt], handler)
}

// Scheduler returns task scheduler for the server
func (server *HTTPServer) Scheduler() tasks.Scheduler {
        return server.scheduler
}

// Service returns a registered server
func (server *HTTPServer) Service(name string) Service {
        server.lock.Lock()
        defer server.lock.Unlock()
        return server.services[name]
}

// HostName returns the host name of the server
func (server *HTTPServer) HostName() string {
        return server.hostname
}

// Port returns the port name of the server
func (server *HTTPServer) Port() string {
        return server.port
}

// Protocol returns the protocol
func (server *HTTPServer) Protocol() string {
        if server.tlsConfig != nil {
                return "https"
        }
        return "http"
}

// LocalIP returns the IP address of the server
func (server *HTTPServer) LocalIP() string {
        return server.ipaddr
}

// StartedAt returns the time when the server started
func (server *HTTPServer) StartedAt() time.Time {
        return server.startedAt
}

// Uptime returns the duration the server was up
func (server *HTTPServer) Uptime() time.Duration {
        return time.Now().UTC().Sub(server.startedAt)
}

// Version returns the version of the server
func (server *HTTPServer) Version() string {
        return server.version
}

// Name returns the server name
func (server *HTTPServer) Name() string {
        return server.httpConfig.GetServiceName()
}

// HTTPConfig returns HTTPServerConfig
func (server *HTTPServer) HTTPConfig() HTTPServerConfig {
        return server.httpConfig
}

// TLSConfig returns TLSConfig
func (server *HTTPServer) TLSConfig() *tls.Config {
        return server.tlsConfig
}

// IsReady returns true when the server is ready to serve
func (server *HTTPServer) IsReady() bool {
        if !server.serving {
                return false
        }
        for _, ss := range server.services {
                if !ss.IsReady() {
                        return false
                }
        }
        return true
}

// Audit create an audit event
func (server *HTTPServer) Audit(source string,
        eventType string,
        identity string,
        contextID string,
        raftIndex uint64,
        message string) {
        if server.auditor != nil {
                server.auditor.Audit(source, eventType, identity, contextID, raftIndex, message)
        } else {
                // {contextID}:{identity}:{raftIndex}:{source}:{type}:{message}
                logger.Infof("audit:%s:%s:%s:%s:%d:%s\n",
                        source, eventType, identity, contextID, raftIndex, message)
        }
}

// WithMuxFactory requires the server to use `muxFactory` to create server handler.
func (server *HTTPServer) WithMuxFactory(muxFactory MuxFactory) {
        server.muxFactory = muxFactory
}

// StartHTTP will verify all the TLS related files are present and start the actual HTTPS listener for the server
func (server *HTTPServer) StartHTTP() error {
        bindAddr := server.httpConfig.GetBindAddr()
        var err error

        // Main server
        if _, err = net.ResolveTCPAddr("tcp", bindAddr); err != nil {
                return errors.WithMessagef(err, "reason=ResolveTCPAddr, service=%s, bind=%q",
                        server.Name(), bindAddr)
        }

        server.httpServer = &http.Server{
                IdleTimeout: time.Hour * 2,
                ErrorLog:    xlog.Stderr,
        }

        var httpsListener net.Listener

        if server.tlsConfig != nil {
                // Start listening on main server over TLS
                httpsListener, err = tls.Listen("tcp", bindAddr, server.tlsConfig)
                if err != nil {
                        return errors.WithMessagef(err, "reason=unable_listen, service=%s, address=%q",
                                server.Name(), bindAddr)
                }

                server.httpServer.TLSConfig = server.tlsConfig
        } else {
                server.httpServer.Addr = bindAddr
        }

        httpHandler := server.muxFactory.NewMux()

        if server.httpConfig.GetAllowProfiling() {
                if httpHandler, err = xhttp.NewRequestProfiler(httpHandler, server.httpConfig.GetProfilerDir(), nil, xhttp.LogProfile()); err != nil {
                        return errors.WithStack(err)
                }
        }

        server.httpServer.Handler = httpHandler

        serve := func() error {
                server.serving = true
                if httpsListener != nil {
                        return server.httpServer.Serve(httpsListener)
                }
                return server.httpServer.ListenAndServe()
        }

        go func() {
                for _, handler := range server.evtHandlers[ServerStartedEvent] {
                        handler(ServerStartedEvent)
                }

                logger.Infof("service=%s, port=%v, status=starting, protocol=%s",
                        server.Name(), bindAddr, server.Protocol())

                // this is a blocking call to serve
                if err := serve(); err != nil {
                        server.serving = false
                        //panic, only if not Serve error while stopping the server,
                        // which is a valid error
                        if netutil.IsAddrInUse(err) || err != http.ErrServerClosed {
                                logger.Panicf("service=%s, err=[%v]", server.Name(), errors.WithStack(err))
                        }
                        logger.Warningf("service=%s, status=stopped, reason=[%s]", server.Name(), err.Error())
                }
        }()

        if server.Scheduler() != nil {
                if server.httpConfig.GetHeartbeatSecs() > 0 {
                        task := tasks.NewTaskAtIntervals(uint64(server.httpConfig.GetHeartbeatSecs()), tasks.Seconds).
                                Do("hearbeat", hearbeatMetricsTask, server)
                        server.Scheduler().Add(task)
                        task.Run()
                }
        }

        server.Audit(
                EvtSourceStatus,
                EvtServiceStarted,
                server.HostName(),
                server.LocalIP(),
                0,
                fmt.Sprintf("address=%q, ClientAuth=%s",
                        strings.TrimPrefix(bindAddr, ":"), server.clientAuth),
        )

        return nil
}

func hearbeatMetricsTask(server *HTTPServer) {
        metricsutil.PublishHeartbeat(server.httpConfig.GetServiceName())
        metricsutil.PublishUptime(server.httpConfig.GetServiceName(), server.Uptime())
}

// StopHTTP will perform a graceful shutdown of the serivce by
//                1) signally to the Load Balancer to remove this instance from the pool
//                                by changing to response to /availability
//                2) cause new responses to have their Connection closed when finished
//                                to force clients to re-connect [hopefully to a different instance]
//                3) wait the minShutdownTime to ensure the LB has noticed the status change
//                4) wait for existing requests to finish processing
//                5) step 4 is capped by a overrall timeout where we'll give up waiting
//                         for the requests to complete and will exit.
//
// it is expected that you don't try and use the server instance again
// after this. [i.e. if you want to start it again, create another server instance]
func (server *HTTPServer) StopHTTP() {
        // close services
        for _, f := range server.services {
                logger.Tracef("service=%q", f.Name())
                f.Close()
        }

        ctx, cancel := context.WithTimeout(context.Background(), server.shutdownTimeout)
        defer cancel()
        err := server.httpServer.Shutdown(ctx)
        if err != nil {
                logger.Errorf("reason=Shutdown, err=[%+v]", err)
        }

        for _, handler := range server.evtHandlers[ServerStoppedEvent] {
                handler(ServerStoppedEvent)
        }

        ut := server.Uptime() / time.Second * time.Second
        server.Audit(
                EvtSourceStatus,
                EvtServiceStopped,
                server.HostName(),
                server.LocalIP(),
                0,
                fmt.Sprintf("uptime=%s", ut),
        )
}

// NewMux creates a new http handler for the http server, typically you only
// need to call this directly for tests.
func (server *HTTPServer) NewMux() http.Handler {
        var router Router
        if server.cors != nil {
                router = NewRouterWithCORS(notFoundHandler, server.cors)
        } else {
                router = NewRouter(notFoundHandler)
        }

        for _, f := range server.services {
                f.Register(router)
        }
        logger.Debugf("service=%s, service_count=%d",
                server.Name(), len(server.services))

        var err error
        httpHandler := router.Handler()

        logger.Infof("service=%s, ClientAuth=%s", server.Name(), server.clientAuth)

        // service ready
        httpHandler = ready.NewServiceStatusVerifier(server, httpHandler)

        if server.authz != nil {
                httpHandler, err = server.authz.NewHandler(httpHandler)
                if err != nil {
                        panic(fmt.Sprintf("%+v", err))
                }
        }

        // logging wrapper
        httpHandler = xhttp.NewRequestLogger(httpHandler, server.Name(), serverExtraLogger, time.Millisecond, server.httpConfig.GetPackageLogger())

        // metrics wrapper
        httpHandler = xhttp.NewRequestMetrics(httpHandler)

        // role/contextID wrapper
        if server.identityMapper != nil {
                httpHandler = identity.NewContextHandler(httpHandler, server.identityMapper)
        } else {
                httpHandler = identity.NewContextHandler(httpHandler, identity.GuestIdentityMapper)
        }
        return httpHandler
}

// ServeHTTP should write reply headers and data to the ResponseWriter
// and then return. Returning signals that the request is finished; it
// is not valid to use the ResponseWriter or read from the
// Request.Body after or concurrently with the completion of the
// ServeHTTP call.
func (server *HTTPServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
        server.httpServer.Handler.ServeHTTP(w, r)
}

func notFoundHandler(w http.ResponseWriter, r *http.Request) {
        marshal.WriteJSON(w, r, httperror.WithNotFound(r.URL.Path))
}

func serverExtraLogger(resp *xhttp.ResponseCapture, req *http.Request) []string {
        return []string{identity.FromRequest(req).CorrelationID()}
}

// GetServerURL returns complete server URL for given relative end-point
func GetServerURL(s Server, r *http.Request, relativeEndpoint string) *url.URL {
        proto := s.Protocol()

        // Allow upstream proxies  to specify the forwarded protocol. Allow this value
        // to override our own guess.
        if specifiedProto := r.Header.Get(header.XForwardedProto); specifiedProto != "" {
                proto = specifiedProto
        }

        host := r.URL.Host
        if host == "" {
                host = r.Host
        }
        if host == "" {
                host = s.HostName() + ":" + s.Port()
        }

        return &url.URL{
                Scheme: proto,
                Host:   host,
                Path:   relativeEndpoint,
        }
}

// GetServerBaseURL returns server base URL
func GetServerBaseURL(s Server) *url.URL {
        return &url.URL{
                Scheme: s.Protocol(),
                Host:   s.HostName() + ":" + s.Port(),
        }
}

1	package rest
2
3	import (
4	"context"
5	"crypto/tls"
6	"fmt"
7	"net"
8	"net/http"
9	"net/url"
10	"strings"
11	"sync"
12	"time"
13
14	metricsutil "github.com/go-phorce/dolly/metrics/util"
15	"github.com/go-phorce/dolly/netutil"
16	"github.com/go-phorce/dolly/rest/ready"
17	"github.com/go-phorce/dolly/tasks"
18	"github.com/go-phorce/dolly/xhttp"
19	"github.com/go-phorce/dolly/xhttp/authz"
20	"github.com/go-phorce/dolly/xhttp/header"
21	"github.com/go-phorce/dolly/xhttp/httperror"
22	"github.com/go-phorce/dolly/xhttp/identity"
23	"github.com/go-phorce/dolly/xhttp/marshal"
24	"github.com/go-phorce/dolly/xlog"
25	"github.com/pkg/errors"
26	)
27
28	var logger = xlog.NewPackageLogger("github.com/go-phorce/dolly", "rest")
29
30	// MaxRequestSize specifies max size of regular HTTP Post requests in bytes, 64 Mb
31	const MaxRequestSize = 64 * 1024 * 1024
32
33	const (
34	// EvtSourceStatus specifies source for service Status
35	EvtSourceStatus = "status"
36	// EvtServiceStarted specifies Service Started event
37	EvtServiceStarted = "service started"
38	// EvtServiceStopped specifies Service Stopped event
39	EvtServiceStopped = "service stopped"
40	)
41
42	// ServerEvent specifies server event type
43	type ServerEvent int
44
45	const (
46	// ServerStartedEvent is fired on server start
47	ServerStartedEvent ServerEvent = iota
48	// ServerStoppedEvent is fired after server stopped
49	ServerStoppedEvent
50	// ServerStoppingEvent is fired before server stopped
51	ServerStoppingEvent
52	)
53
54	// ServerEventFunc is a callback to handle server events
55	type ServerEventFunc func(evt ServerEvent)
56
57	// Server is an interface to provide server status
58	type Server interface {
59	http.Handler
60	Name() string
61	Version() string
62	HostName() string
63	LocalIP() string
64	Port() string
65	Protocol() string
66	StartedAt() time.Time
67	Uptime() time.Duration
68	Service(name string) Service
69	HTTPConfig() HTTPServerConfig
70	TLSConfig() *tls.Config
71
72	// IsReady indicates that all subservices are ready to serve
73	IsReady() bool
74
75	// Audit records an auditable event.
76	// source indicates the area that the event was triggered by
77	// eventType indicates the specific event that occured
78	// identity specifies the identity of the user that triggered this event, typically this is <role>/<cn>
79	// contextID specifies the request ContextID that the event was triggered in [this can be used for cross service correlation of logs]
80	// raftIndex indicates the index# of the raft log in RAFT that the event occured in [if applicable]
81	// message contains any additional information about this event that is eventType specific
82	Audit(source string,
83	eventType string,
84	identity string,
85	contextID string,
86	raftIndex uint64,
87	message string)
88
89	AddService(s Service)
90	StartHTTP() error
91	StopHTTP()
92
93	Scheduler() tasks.Scheduler
94
95	OnEvent(evt ServerEvent, handler ServerEventFunc)
96	}
97
98	// MuxFactory creates http handlers.
99	type MuxFactory interface {
100	NewMux() http.Handler
101	}
102
103	// HTTPServer is responsible for exposing the collection of the services
104	// as a single HTTP server
105	type HTTPServer struct {
106	Server
107	auditor Auditor
108	authz authz.HTTPAuthz
109	identityMapper identity.ProviderFromRequest
110	httpConfig HTTPServerConfig
111	tlsConfig *tls.Config
112	httpServer *http.Server
113	cors *CORSOptions
114	muxFactory MuxFactory
115	hostname string
116	port string
117	ipaddr string
118	version string
119	serving bool
120	startedAt time.Time
121	clientAuth string
122	scheduler tasks.Scheduler
123	services map[string]Service
124	evtHandlers map[ServerEvent][]ServerEventFunc
125	lock sync.RWMutex
126	shutdownTimeout time.Duration
127	}
128
129	// New creates a new instance of the server
130	func New(
131	version string,
132	ipaddr string,
133	httpConfig HTTPServerConfig,
134	tlsConfig *tls.Config,
135	) (*HTTPServer, error) {	16✔
136	var err error	16✔
137		16✔
138	if ipaddr == "" {	24✔
139	ipaddr, err = netutil.GetLocalIP()	8✔
140	if err != nil {	8✔
141	ipaddr = "127.0.0.1"	×
142	logger.Errorf("reason=unable_determine_ipaddr, use=%q, err=[%+v]", ipaddr, err)	×
143	}	×
144	}
145
146	s := &HTTPServer{	16✔
147	services: map[string]Service{},	16✔
148	startedAt: time.Now().UTC(),	16✔
149	version: version,	16✔
150	ipaddr: ipaddr,	16✔
151	evtHandlers: make(map[ServerEvent][]ServerEventFunc),	16✔
152	clientAuth: tlsClientAuthToStrMap[tls.NoClientCert],	16✔
153	httpConfig: httpConfig,	16✔
154	hostname: GetHostName(httpConfig.GetBindAddr()),	16✔
155	port: GetPort(httpConfig.GetBindAddr()),	16✔
156	tlsConfig: tlsConfig,	16✔
157	shutdownTimeout: time.Duration(5) * time.Second,	16✔
158	}	16✔
159	s.muxFactory = s	16✔
160	if tlsConfig != nil {	25✔
161	s.clientAuth = tlsClientAuthToStrMap[tlsConfig.ClientAuth]	9✔
162	}	9✔
163
164	return s, nil	16✔
165	}
166
167	// WithAuditor enables to use auditor
168	func (server HTTPServer) WithAuditor(auditor Auditor) HTTPServer {	6✔
169	server.auditor = auditor	6✔
170	return server	6✔
171	}	6✔
172
173	// WithAuthz enables to use Authz
174	func (server HTTPServer) WithAuthz(authz authz.HTTPAuthz) HTTPServer {	5✔
175	server.authz = authz	5✔
176	return server	5✔
177	}	5✔
178
179	// WithIdentityProvider enables to set idenity on each request
180	func (server HTTPServer) WithIdentityProvider(provider identity.ProviderFromRequest) HTTPServer {	2✔
181	server.identityMapper = provider	2✔
182	return server	2✔
183	}	2✔
184
185	// WithScheduler enables to schedule tasks, such as heartbeat, uptime etc
186	// If a scheduler is not provided, then tasks will not be schedduled.
187	func (server HTTPServer) WithScheduler(scheduler tasks.Scheduler) HTTPServer {	1✔
188	server.scheduler = scheduler	1✔
189	return server	1✔
190	}	1✔
191
192	// WithCORS enables CORS options
193	func (server HTTPServer) WithCORS(cors CORSOptions) *HTTPServer {	1✔
194	server.cors = cors	1✔
195	return server	1✔
196	}	1✔
197
198	// WithShutdownTimeout sets the connection draining timeouts on server shutdown
199	func (server HTTPServer) WithShutdownTimeout(timeout time.Duration) HTTPServer {	1✔
200	server.shutdownTimeout = timeout	1✔
201	return server	1✔
202	}	1✔
203
204	var tlsClientAuthToStrMap = map[tls.ClientAuthType]string{
205	tls.NoClientCert: "NoClientCert",
206	tls.RequestClientCert: "RequestClientCert",
207	tls.RequireAnyClientCert: "RequireAnyClientCert",
208	tls.VerifyClientCertIfGiven: "VerifyClientCertIfGiven",
209	tls.RequireAndVerifyClientCert: "RequireAndVerifyClientCert",
210	}
211
212	// AddService provides a service registration for the server
213	func (server *HTTPServer) AddService(s Service) {	11✔
214	server.lock.Lock()	11✔
215	defer server.lock.Unlock()	11✔
216	server.services[s.Name()] = s	11✔
217	}	11✔
218
219	// OnEvent accepts a callback to handle server events
220	func (server *HTTPServer) OnEvent(evt ServerEvent, handler ServerEventFunc) {	4✔
221	server.lock.Lock()	4✔
222	defer server.lock.Unlock()	4✔
223		4✔
224	server.evtHandlers[evt] = append(server.evtHandlers[evt], handler)	4✔
225	}	4✔
226
227	// Scheduler returns task scheduler for the server
228	func (server *HTTPServer) Scheduler() tasks.Scheduler {	16✔
229	return server.scheduler	16✔
230	}	16✔
231
232	// Service returns a registered server
233	func (server *HTTPServer) Service(name string) Service {	2✔
234	server.lock.Lock()	2✔
235	defer server.lock.Unlock()	2✔
236	return server.services[name]	2✔
237	}	2✔
238
239	// HostName returns the host name of the server
240	func (server *HTTPServer) HostName() string {	35✔
241	return server.hostname	35✔
242	}	35✔
243
244	// Port returns the port name of the server
245	func (server *HTTPServer) Port() string {	15✔
246	return server.port	15✔
247	}	15✔
248
249	// Protocol returns the protocol
250	func (server *HTTPServer) Protocol() string {	31✔
251	if server.tlsConfig != nil {	47✔
252	return "https"	16✔
253	}	16✔
254	return "http"	15✔
255	}
256
257	// LocalIP returns the IP address of the server
258	func (server *HTTPServer) LocalIP() string {	28✔
259	return server.ipaddr	28✔
260	}	28✔
261
262	// StartedAt returns the time when the server started
263	func (server *HTTPServer) StartedAt() time.Time {	2✔
264	return server.startedAt	2✔
265	}	2✔
266
267	// Uptime returns the duration the server was up
268	func (server *HTTPServer) Uptime() time.Duration {	13✔
269	return time.Now().UTC().Sub(server.startedAt)	13✔
270	}	13✔
271
272	// Version returns the version of the server
273	func (server *HTTPServer) Version() string {	2✔
274	return server.version	2✔
275	}	2✔
276
277	// Name returns the server name
278	func (server *HTTPServer) Name() string {	66✔
279	return server.httpConfig.GetServiceName()	66✔
280	}	66✔
281
282	// HTTPConfig returns HTTPServerConfig
283	func (server *HTTPServer) HTTPConfig() HTTPServerConfig {	4✔
284	return server.httpConfig	4✔
285	}	4✔
286
287	// TLSConfig returns TLSConfig
288	func (server HTTPServer) TLSConfig() tls.Config {	1✔
289	return server.tlsConfig	1✔
290	}	1✔
291
292	// IsReady returns true when the server is ready to serve
293	func (server *HTTPServer) IsReady() bool {	50✔
294	if !server.serving {	64✔
295	return false	14✔
296	}	14✔
297	for _, ss := range server.services {	68✔
298	if !ss.IsReady() {	32✔
299	return false	×
300	}	×
301	}
302	return true	36✔
303	}
304
305	// Audit create an audit event
306	func (server *HTTPServer) Audit(source string,
307	eventType string,
308	identity string,
309	contextID string,
310	raftIndex uint64,
311	message string) {	26✔
312	if server.auditor != nil {	36✔
313	server.auditor.Audit(source, eventType, identity, contextID, raftIndex, message)	10✔
314	} else {	26✔
315	// {contextID}:{identity}:{raftIndex}:{source}:{type}:{message}	16✔
316	logger.Infof("audit:%s:%s:%s:%s:%d:%s\n",	16✔
317	source, eventType, identity, contextID, raftIndex, message)	16✔
318	}	16✔
319	}
320
321	// WithMuxFactory requires the server to use `muxFactory` to create server handler.
322	func (server *HTTPServer) WithMuxFactory(muxFactory MuxFactory) {	1✔
323	server.muxFactory = muxFactory	1✔
324	}	1✔
325
326	// StartHTTP will verify all the TLS related files are present and start the actual HTTPS listener for the server
327	func (server *HTTPServer) StartHTTP() error {	14✔
328	bindAddr := server.httpConfig.GetBindAddr()	14✔
329	var err error	14✔
330		14✔
331	// Main server	14✔
332	if _, err = net.ResolveTCPAddr("tcp", bindAddr); err != nil {	15✔
333	return errors.WithMessagef(err, "reason=ResolveTCPAddr, service=%s, bind=%q",	1✔
334	server.Name(), bindAddr)	1✔
335	}	1✔
336
337	server.httpServer = &http.Server{	13✔
338	IdleTimeout: time.Hour * 2,	13✔
339	ErrorLog: xlog.Stderr,	13✔
340	}	13✔
341		13✔
342	var httpsListener net.Listener	13✔
343		13✔
344	if server.tlsConfig != nil {	21✔
345	// Start listening on main server over TLS	8✔
346	httpsListener, err = tls.Listen("tcp", bindAddr, server.tlsConfig)	8✔
347	if err != nil {	8✔
348	return errors.WithMessagef(err, "reason=unable_listen, service=%s, address=%q",	×
349	server.Name(), bindAddr)	×
350	}	×
351
352	server.httpServer.TLSConfig = server.tlsConfig	8✔
353	} else {	5✔
354	server.httpServer.Addr = bindAddr	5✔
355	}	5✔
356
357	httpHandler := server.muxFactory.NewMux()	13✔
358		13✔
359	if server.httpConfig.GetAllowProfiling() {	13✔
360	if httpHandler, err = xhttp.NewRequestProfiler(httpHandler, server.httpConfig.GetProfilerDir(), nil, xhttp.LogProfile()); err != nil {	×
361	return errors.WithStack(err)	×
362	}	×
363	}
364
365	server.httpServer.Handler = httpHandler	13✔
366		13✔
367	serve := func() error {	26✔
368	server.serving = true	13✔
369	if httpsListener != nil {	21✔
370	return server.httpServer.Serve(httpsListener)	8✔
371	}	8✔
372	return server.httpServer.ListenAndServe()	5✔
373	}
374
375	go func() {	26✔
376	for _, handler := range server.evtHandlers[ServerStartedEvent] {	15✔
377	handler(ServerStartedEvent)	2✔
378	}	2✔
379
380	logger.Infof("service=%s, port=%v, status=starting, protocol=%s",	13✔
381	server.Name(), bindAddr, server.Protocol())	13✔
382		13✔
383	// this is a blocking call to serve	13✔
384	if err := serve(); err != nil {	26✔
385	server.serving = false	13✔
386	//panic, only if not Serve error while stopping the server,	13✔
387	// which is a valid error	13✔
388	if netutil.IsAddrInUse(err) \|\| err != http.ErrServerClosed {	13✔
389	logger.Panicf("service=%s, err=[%v]", server.Name(), errors.WithStack(err))	×
390	}	×
391	logger.Warningf("service=%s, status=stopped, reason=[%s]", server.Name(), err.Error())	13✔
392	}
393	}()
394
395	if server.Scheduler() != nil {	14✔
396	if server.httpConfig.GetHeartbeatSecs() > 0 {	1✔
397	task := tasks.NewTaskAtIntervals(uint64(server.httpConfig.GetHeartbeatSecs()), tasks.Seconds).	×
398	Do("hearbeat", hearbeatMetricsTask, server)	×
399	server.Scheduler().Add(task)	×
400	task.Run()	×
401	}	×
402	}
403
404	server.Audit(	13✔
405	EvtSourceStatus,	13✔
406	EvtServiceStarted,	13✔
407	server.HostName(),	13✔
408	server.LocalIP(),	13✔
409	0,	13✔
410	fmt.Sprintf("address=%q, ClientAuth=%s",	13✔
411	strings.TrimPrefix(bindAddr, ":"), server.clientAuth),	13✔
412	)	13✔
413		13✔
414	return nil	13✔
415	}
416
417	func hearbeatMetricsTask(server *HTTPServer) {	×
418	metricsutil.PublishHeartbeat(server.httpConfig.GetServiceName())	×
419	metricsutil.PublishUptime(server.httpConfig.GetServiceName(), server.Uptime())	×
420	}	×
421
422	// StopHTTP will perform a graceful shutdown of the serivce by
423	// 1) signally to the Load Balancer to remove this instance from the pool
424	// by changing to response to /availability
425	// 2) cause new responses to have their Connection closed when finished
426	// to force clients to re-connect [hopefully to a different instance]
427	// 3) wait the minShutdownTime to ensure the LB has noticed the status change
428	// 4) wait for existing requests to finish processing
429	// 5) step 4 is capped by a overrall timeout where we'll give up waiting
430	// for the requests to complete and will exit.
431	//
432	// it is expected that you don't try and use the server instance again
433	// after this. [i.e. if you want to start it again, create another server instance]
434	func (server *HTTPServer) StopHTTP() {	13✔
435	// close services	13✔
436	for _, f := range server.services {	24✔
437	logger.Tracef("service=%q", f.Name())	11✔
438	f.Close()	11✔
439	}	11✔
440
441	ctx, cancel := context.WithTimeout(context.Background(), server.shutdownTimeout)	13✔
442	defer cancel()	13✔
443	err := server.httpServer.Shutdown(ctx)	13✔
444	if err != nil {	13✔
445	logger.Errorf("reason=Shutdown, err=[%+v]", err)	×
446	}	×
447
448	for _, handler := range server.evtHandlers[ServerStoppedEvent] {	15✔
449	handler(ServerStoppedEvent)	2✔
450	}	2✔
451
452	ut := server.Uptime() / time.Second * time.Second	13✔
453	server.Audit(	13✔
454	EvtSourceStatus,	13✔
455	EvtServiceStopped,	13✔
456	server.HostName(),	13✔
457	server.LocalIP(),	13✔
458	0,	13✔
459	fmt.Sprintf("uptime=%s", ut),	13✔
460	)	13✔
461	}
462
463	// NewMux creates a new http handler for the http server, typically you only
464	// need to call this directly for tests.
465	func (server *HTTPServer) NewMux() http.Handler {	13✔
466	var router Router	13✔
467	if server.cors != nil {	14✔
468	router = NewRouterWithCORS(notFoundHandler, server.cors)	1✔
469	} else {	13✔
470	router = NewRouter(notFoundHandler)	12✔
471	}	12✔
472
473	for _, f := range server.services {	24✔
474	f.Register(router)	11✔
475	}	11✔
476	logger.Debugf("service=%s, service_count=%d",	13✔
477	server.Name(), len(server.services))	13✔
478		13✔
479	var err error	13✔
480	httpHandler := router.Handler()	13✔
481		13✔
482	logger.Infof("service=%s, ClientAuth=%s", server.Name(), server.clientAuth)	13✔
483		13✔
484	// service ready	13✔
485	httpHandler = ready.NewServiceStatusVerifier(server, httpHandler)	13✔
486		13✔
487	if server.authz != nil {	18✔
488	httpHandler, err = server.authz.NewHandler(httpHandler)	5✔
489	if err != nil {	5✔
490	panic(fmt.Sprintf("%+v", err))	×
491	}
492	}
493
494	// logging wrapper
495	httpHandler = xhttp.NewRequestLogger(httpHandler, server.Name(), serverExtraLogger, time.Millisecond, server.httpConfig.GetPackageLogger())	13✔
496		13✔
497	// metrics wrapper	13✔
498	httpHandler = xhttp.NewRequestMetrics(httpHandler)	13✔
499		13✔
500	// role/contextID wrapper	13✔
501	if server.identityMapper != nil {	15✔
502	httpHandler = identity.NewContextHandler(httpHandler, server.identityMapper)	2✔
503	} else {	13✔
504	httpHandler = identity.NewContextHandler(httpHandler, identity.GuestIdentityMapper)	11✔
505	}	11✔
506	return httpHandler	13✔
507	}
508
509	// ServeHTTP should write reply headers and data to the ResponseWriter
510	// and then return. Returning signals that the request is finished; it
511	// is not valid to use the ResponseWriter or read from the
512	// Request.Body after or concurrently with the completion of the
513	// ServeHTTP call.
514	func (server HTTPServer) ServeHTTP(w http.ResponseWriter, r http.Request) {	11✔
515	server.httpServer.Handler.ServeHTTP(w, r)	11✔
516	}	11✔
517
518	func notFoundHandler(w http.ResponseWriter, r *http.Request) {	1✔
519	marshal.WriteJSON(w, r, httperror.WithNotFound(r.URL.Path))	1✔
520	}	1✔
521
522	func serverExtraLogger(resp xhttp.ResponseCapture, req http.Request) []string {	14✔
523	return []string{identity.FromRequest(req).CorrelationID()}	14✔
524	}	14✔
525
526	// GetServerURL returns complete server URL for given relative end-point
527	func GetServerURL(s Server, r http.Request, relativeEndpoint string) url.URL {	3✔
528	proto := s.Protocol()	3✔
529		3✔
530	// Allow upstream proxies to specify the forwarded protocol. Allow this value	3✔
531	// to override our own guess.	3✔
532	if specifiedProto := r.Header.Get(header.XForwardedProto); specifiedProto != "" {	5✔
533	proto = specifiedProto	2✔
534	}	2✔
535
536	host := r.URL.Host	3✔
537	if host == "" {	6✔
538	host = r.Host	3✔
539	}	3✔
540	if host == "" {	5✔
541	host = s.HostName() + ":" + s.Port()	2✔
542	}	2✔
543
544	return &url.URL{	3✔
545	Scheme: proto,	3✔
546	Host: host,	3✔
547	Path: relativeEndpoint,	3✔
548	}	3✔
549	}
550
551	// GetServerBaseURL returns server base URL
552	func GetServerBaseURL(s Server) *url.URL {	2✔
553	return &url.URL{	2✔
554	Scheme: s.Protocol(),	2✔
555	Host: s.HostName() + ":" + s.Port(),	2✔
556	}	2✔
557	}	2✔

go-phorce / dolly / 4350203089

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous