http_api_server.go ×23

Frontier kind: Code frontier

unlabeled · c_b5921ddb300a

5 tests · 26194 LOC · 626 files · introduces 0 tests · 195 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
31 ranges195 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
5420 ranges26194 lines · 626 files · Browse complete extent
All tests (intent)
5 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

5 files ranked by introduced lines: 195 introduced LOC across 31 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/frontend/http_api_server.go 144 introduced LOC · 23 ranges

Open complete file

79 namespaceRegistry namespace.Registry,
80 logger log.Logger,
81 > ) (*HTTPAPIServer, error) { http_api_server.go
82 > // Create a TCP listener the same as the frontend one but with different port
83 > tcpAddrRef, _ := grpcListener.Addr().(*net.TCPAddr)
84 > if tcpAddrRef == nil {
85 return nil, errHTTPGRPCListenerNotTCP
86 }
87 > tcpAddr := *tcpAddrRef http_api_server.go
88 > tcpAddr.Port = rpcConfig.HTTPPort
89 > var listener net.Listener
90 > var err error
91 > if listener, err = net.ListenTCP("tcp", &tcpAddr); err != nil {
92 return nil, fmt.Errorf("failed listening for HTTP API on %v: %w", &tcpAddr, err)
93 }
94 // Close the listener if anything else in this function fails
95 > success := false http_api_server.go
96 > defer func() {
97 > if !success {
98 _ = listener.Close()
99 }
101
102 // Wrap the listener in a TLS listener if there is any TLS config
103 > if tlsConfigProvider != nil { http_api_server.go
104 if tlsConfig, err := tlsConfigProvider.GetFrontendServerConfig(); err != nil {
105 return nil, fmt.Errorf("failed getting TLS config for HTTP API: %w", err)
109 }
110
111 > h := &HTTPAPIServer{ http_api_server.go
112 > listener: listener,
113 > logger: logger,
114 > stopped: make(chan struct{}),
115 > allowedHosts: serviceConfig.HTTPAllowedHosts,
116 > }
117 >
118 > // Build 4 possible marshalers in order based on content type
119 > opts := []runtime.ServeMuxOption{
120 > runtime.WithMarshalerOption(newTemporalProtoMarshaler(" ", false)),
121 > runtime.WithMarshalerOption(newTemporalProtoMarshaler("", false)),
122 > runtime.WithMarshalerOption(newTemporalProtoMarshaler(" ", true)),
123 > runtime.WithMarshalerOption(newTemporalProtoMarshaler("", true)),
124 > }
125 >
126 > // Set Temporal service error handler
127 > opts = append(opts, runtime.WithErrorHandler(h.errorHandler))
128 >
129 > // Match headers w/ default
130 > h.matchAdditionalHeaders = map[string]bool{}
131 > for _, v := range defaultForwardedHeaders {
132 > h.matchAdditionalHeaders[v] = true
133 > }
134 > for _, v := range rpcConfig.HTTPAdditionalForwardedHeaders {
135 if before, ok := strings.CutSuffix(v, "*"); ok {
136 h.matchAdditionalHeaderPrefixes = append(h.matchAdditionalHeaderPrefixes, http.CanonicalHeaderKey(before))
140 }
141
142 > opts = append(opts, runtime.WithMiddlewares(h.allowedHostsMiddleware)) http_api_server.go
143 > opts = append(opts, runtime.WithIncomingHeaderMatcher(h.incomingHeaderMatcher))
144 >
145 > // Create inline client connection
146 > clientConn := newInlineClientConn(
147 > map[string]any{
148 > "temporal.api.workflowservice.v1.WorkflowService": handler,
149 > "temporal.api.operatorservice.v1.OperatorService": operatorHandler,
150 > },
151 > interceptors,
152 > metricsHandler,
153 > namespaceRegistry,
154 > )
155 >
156 > // Create serve mux
157 > h.serveMux = runtime.NewServeMux(opts...)
158 >
159 > err = workflowservice.RegisterWorkflowServiceHandlerClient(
160 > context.Background(),
161 > h.serveMux,
162 > workflowservice.NewWorkflowServiceClient(clientConn),
163 > )
164 > if err != nil {
165 return nil, fmt.Errorf("failed registering workflowservice HTTP API handler: %w", err)
166 }
167
168 > err = operatorservice.RegisterOperatorServiceHandlerClient( http_api_server.go
169 > context.Background(),
170 > h.serveMux,
171 > operatorservice.NewOperatorServiceClient(clientConn),
172 > )
173 > if err != nil {
174 return nil, fmt.Errorf("failed registering operatorservice HTTP API handler: %w", err)
175 }
176
177 // Set the / handler as our function that wraps serve mux.
178 > router.PathPrefix("/").HandlerFunc(h.serveHTTP) http_api_server.go
179 > // Register the router as the HTTP server handler.
180 > h.server.Handler = router
181 >
182 > // Put the remote address on the context
183 > h.server.ConnContext = func(ctx context.Context, c net.Conn) context.Context {
184 return context.WithValue(ctx, httpRemoteAddrContextKey{}, c)
185 }
188 // defaults to ReadTimeout) to ensure that a connection cannot hang over that
189 // amount of time.
190 > h.server.ReadTimeout = serviceConfig.KeepAliveMaxConnectionIdle() http_api_server.go
191 > h.server.WriteTimeout = serviceConfig.KeepAliveMaxConnectionIdle()
192 >
193 > success = true
194 > return h, nil
195 }
196
198 // GracefulStop completes. Upon graceful stop, this will return nil. If an error
199 // is returned, the message is clear that it came from the HTTP API server.
200 > func (h *HTTPAPIServer) Serve() error { http_api_server.go
201 > err := h.server.Serve(h.listener)
202 > // If the error is for close, we have to wait for the shutdown to complete and
203 > // we don't consider it an error
204 > if errors.Is(err, http.ErrServerClosed) {
205 > <-h.stopped
206 > err = nil
207 > }
208 // Wrap the error to be clearer it's from the HTTP API
209 > if err != nil { http_api_server.go
210 return fmt.Errorf("HTTP API serve failed: %w", err)
211 }
212 > return nil http_api_server.go
213 }
214
215 // GracefulStop stops the HTTP server. This will first attempt a graceful stop
216 // with a drain time, then will hard-stop. This will not return until stopped.
217 > func (h *HTTPAPIServer) GracefulStop(gracefulDrainTime time.Duration) { http_api_server.go
218 > // We try a graceful stop for the amount of time we can drain, then we do a
219 > // hard stop
220 > shutdownCtx, cancel := context.WithTimeout(context.Background(), gracefulDrainTime)
221 > defer cancel()
222 > // We intentionally ignore this error, we're gonna stop at this point no
223 > // matter what. This closes the listener too.
224 > _ = h.server.Shutdown(shutdownCtx)
225 > _ = h.server.Close()
226 > close(h.stopped)
227 > }
228
229 func (h *HTTPAPIServer) serveHTTP(w http.ResponseWriter, r *http.Request) {
272 }
273
274 > func (h *HTTPAPIServer) allowedHostsMiddleware(hf runtime.HandlerFunc) runtime.HandlerFunc { http_api_server.go
275 > return func(w http.ResponseWriter, r *http.Request, pathParams map[string]string) {
276 allowedHosts := h.allowedHosts()
277 if allowedHosts.MatchString(r.Host) {
318 }
319
320 > func (h *HTTPAPIServer) incomingHeaderMatcher(headerName string) (string, bool) { http_api_server.go
321 > // Try ours before falling back to default
322 > if h.matchAdditionalHeaders[headerName] {
323 return headerName, true
324 }
325 > for _, prefix := range h.matchAdditionalHeaderPrefixes { http_api_server.go
326 if strings.HasPrefix(headerName, prefix) {
327 return headerName, true
328 }
329 }
330 > return runtime.DefaultHeaderMatcher(headerName) http_api_server.go
331 }
332
359 metricsHandler metrics.Handler,
360 namespaceRegistry namespace.Registry,
361 > ) *inlineClientConn { http_api_server.go
362 > // Create the set of methods via reflection. We currently accept the overhead
363 > // of reflection compared to having to custom generate gateway code.
364 > methods := map[string]*serviceMethod{}
365 > for qualifiedServerName, server := range servers {
366 > serverVal := reflect.ValueOf(server)
367 > for reflectMethod := range serverVal.Type().Methods() {
368 > // We intentionally look this up by name to not assume method indexes line
369 > // up from type to value
370 > methodVal := serverVal.MethodByName(reflectMethod.Name)
371 > // We assume the methods we want only accept a context + request and only
372 > // return a response + error. We also assume the method name matches the
373 > // RPC name.
374 > methodType := methodVal.Type()
375 > validRPCMethod := methodType.Kind() == reflect.Func &&
376 > methodType.NumIn() == 2 &&
377 > methodType.NumOut() == 2 &&
378 > methodType.In(0) == contextType &&
379 > methodType.In(1).Implements(protoMessageType) &&
380 > methodType.Out(0).Implements(protoMessageType) &&
381 > methodType.Out(1) == errorType
382 > if !validRPCMethod {
383 > continue
384 }
385 > fullMethod := "/" + qualifiedServerName + "/" + reflectMethod.Name http_api_server.go
386 > methods[fullMethod] = &serviceMethod{
387 > info: grpc.UnaryServerInfo{Server: server, FullMethod: fullMethod},
388 > handler: func(ctx context.Context, req any) (any, error) {
389 ret := methodVal.Call([]reflect.Value{reflect.ValueOf(ctx), reflect.ValueOf(req)})
390 err, _ := ret[1].Interface().(error)
395 }
396
397 > return &inlineClientConn{ http_api_server.go
398 > methods: methods,
399 > interceptor: chainUnaryServerInterceptors(interceptors),
400 > requestsCounter: metrics.HTTPServiceRequests.With(metricsHandler),
401 > namespaceRegistry: namespaceRegistry,
402 > }
403 }
404
475 // Mostly taken from https://github.com/grpc/grpc-go/blob/v1.56.1/server.go#L1124-L1158
476 // with slight modifications.
477 > func chainUnaryServerInterceptors(interceptors []grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor { http_api_server.go
478 > switch len(interceptors) {
479 case 0:
480 return nil
481 case 1:
482 return interceptors[0]
483 > default: http_api_server.go
484 > return chainUnaryInterceptors(interceptors)
485 }
486 }
487
488 > func chainUnaryInterceptors(interceptors []grpc.UnaryServerInterceptor) grpc.UnaryServerInterceptor { http_api_server.go
489 > return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
490 return interceptors[0](ctx, req, info, getChainUnaryHandler(interceptors, 0, info, handler))
491 }
go.temporal.io/server/service/frontend/protojson_marshaler.go 26 introduced LOC · 3 ranges

Open complete file

32 }
33
34 > func newTemporalProtoMarshaler(indent string, enablePayloadShorthand bool) (string, temporalProtoMarshaler) { protojson_marshaler.go
35 > metadata := map[string]any{}
36 > if enablePayloadShorthand {
37 > metadata[commonpb.EnablePayloadShorthandMetadataKey] = true
38 > }
39 // Shorthand is enabled by default
40 > contentType := runtime.MIMEWildcard protojson_marshaler.go
41 > if enablePayloadShorthand {
42 > if indent != "" {
43 > contentType = "application/json+pretty"
44 > }
45 > } else {
46 > if indent != "" {
47 > contentType = "application/json+pretty+no-payload-shorthand"
48 > } else {
49 > contentType = "application/json+no-payload-shorthand"
50 > }
51 }
52 > return contentType, temporalProtoMarshaler{ protojson_marshaler.go
53 > contentType: contentType,
54 > mOpts: temporalproto.CustomJSONMarshalOptions{
55 > Indent: indent,
56 > Metadata: metadata,
57 > },
58 > uOpts: temporalproto.CustomJSONUnmarshalOptions{
59 > Metadata: metadata,
60 > },
61 > }
62 }
63
go.temporal.io/server/service/frontend/fx.go 17 introduced LOC · 2 ranges

Open complete file

305 }
306 if len(customInterceptors) > 0 {
307 > // TODO: Deprecate WithChainedFrontendGrpcInterceptors and provide a inner custom interceptor fx.go
308 > unaryInterceptors = append(unaryInterceptors, customInterceptors...)
309 > }
310 // retry interceptor should be the most inner interceptor
311 unaryInterceptors = append(unaryInterceptors, retryableInterceptor.Intercept)
1020 return nil, nil
1021 }
1022 > rpcConfig := cfg.Services[string(serviceName)].RPC fx.go
1023 > return NewHTTPAPIServer(
1024 > serviceConfig,
1025 > rpcConfig,
1026 > grpcListener,
1027 > tlsConfigProvider,
1028 > handler,
1029 > operatorHandler,
1030 > grpcServerOptions.UnaryInterceptors,
1031 > metricsHandler,
1032 > router,
1033 > namespaceRegistry,
1034 > logger,
1035 > )
1036 }
1037
go.temporal.io/server/service/frontend/service.go 5 introduced LOC · 2 ranges

Open complete file

505
506 if s.httpAPIServer != nil {
507 > go func() { service.go
508 > if err := s.httpAPIServer.Serve(); err != nil {
509 s.logger.Fatal("Failed to serve HTTP API server", tag.Error(err))
510 }
555 })
556 if s.httpAPIServer != nil {
557 > wg.Go(func() { service.go
558 > s.httpAPIServer.GracefulStop(requestDrainTime)
559 > })
560 }
561 wg.Wait()
go.temporal.io/server/common/membership/grpc_resolver.go 3 introduced LOC · 1 range

Open complete file

154 }
155
156 > func (m *grpcResolver) ResolveNow(_ resolver.ResolveNowOptions) { grpc_resolver.go
157 > select {
158 > case m.notifyCh <- nil:
159 default:
160 }