@@ -21,16 +21,12 @@ import (
2121 "syscall"
2222 "time"
2323
24- confworkflowtypes "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/actions/confidentialworkflow"
2524 cllogger "github.com/smartcontractkit/chainlink-common/pkg/logger"
26- sdkpb "github.com/smartcontractkit/chainlink-protos/cre/go/sdk"
27- confhttptypes "github.com/smartcontractkit/chainlink-confidential-compute/enclave/apps/confidential-http/types"
2825 signatureverifier "github.com/smartcontractkit/chainlink-confidential-compute/enclave/services/signature-verifier"
2926 "github.com/smartcontractkit/chainlink-confidential-compute/enclave/vsock"
3027 "github.com/smartcontractkit/chainlink-confidential-compute/types"
3128 "github.com/smartcontractkit/chainlink-confidential-compute/util"
3229 "go.uber.org/zap/zapcore"
33- "google.golang.org/protobuf/proto"
3430)
3531
3632var (
@@ -40,11 +36,12 @@ var (
4036)
4137
4238var (
43- httpPort = flag .Int ("port" , 8080 , "HTTP port to listen on" )
44- configHttpPort = flag .Int ("config-port" , 8081 , "HTTP port for config endpoint (localhost only)" )
45- enclavePort = flag .Int ("enclave-port" , 5000 , "VSOCK port the enclave is listening on" )
46- enclaveCID = flag .Int ("enclave-cid" , 16 , "VSOCK CID of the enclave" )
47- quorumTimeout = flag .Duration ("quorum-timeout" , types .QuorumTimeout , "Timeout for waiting for quorum to be reached" )
39+ httpPort = flag .Int ("port" , 8080 , "HTTP port to listen on" )
40+ configHttpPort = flag .Int ("config-port" , 8081 , "HTTP port for config endpoint (localhost only)" )
41+ enclavePort = flag .Int ("enclave-port" , 5000 , "VSOCK port the enclave is listening on" )
42+ enclaveCID = flag .Int ("enclave-cid" , 16 , "VSOCK CID of the enclave" )
43+ quorumTimeout = flag .Duration ("quorum-timeout" , types .QuorumTimeout , "Timeout for waiting for quorum to be reached" )
44+ shutdownTimeout = flag .Duration ("shutdown-timeout" , 25 * time .Second , "Maximum time to drain servers and flush telemetry during shutdown; keep below the process termination grace period" )
4845 // requireBFTQuorum raises the batch quorum threshold from f+1 (one honest
4946 // node) to 2f+1 (a BFT supermajority). Reads REQUIRE_BFT_QUORUM.
5047 requireBFTQuorum = flag .Bool ("require-bft-quorum" , os .Getenv ("REQUIRE_BFT_QUORUM" ) == "true" , "require a 2f+1 BFT quorum instead of f+1. Reads REQUIRE_BFT_QUORUM." )
@@ -111,6 +108,7 @@ type hostServer struct {
111108 responseCache * util.Cache [* types.ExecuteResponse ]
112109 verifier signatureverifier.SignatureVerifier
113110 logger cllogger.SugaredLogger
111+ metrics executionMetrics
114112}
115113
116114type batchRequest struct {
@@ -134,108 +132,6 @@ type batchResponse struct {
134132 err error
135133}
136134
137- func logPublicData (reqLog cllogger.SugaredLogger , appID string , publicData []byte ) {
138- switch appID {
139- case types .AppIDConfidentialHTTP :
140- var req confhttptypes.Request
141- if err := proto .Unmarshal (publicData , & req ); err != nil {
142- reqLog .Warnw ("failed to decode publicData" ,
143- "event" , "PUBLIC_DATA_DECODE_ERR" ,
144- "appID" , appID ,
145- "publicDataLen" , len (publicData ),
146- "error" , err )
147- return
148- }
149-
150- bodyKind := "none"
151- bodyLen := 0
152- switch body := req .GetBody ().(type ) {
153- case * confhttptypes.Request_BodyString :
154- bodyKind = "string"
155- bodyLen = len (body .BodyString )
156- case * confhttptypes.Request_BodyBytes :
157- bodyKind = "bytes"
158- bodyLen = len (body .BodyBytes )
159- }
160-
161- timeout := ""
162- if req .GetTimeout () != nil {
163- timeout = req .GetTimeout ().AsDuration ().String ()
164- }
165-
166- reqLog .Infow ("decoded publicData" ,
167- "event" , "PUBLIC_DATA" ,
168- "appID" , appID ,
169- "publicDataLen" , len (publicData ),
170- "publicDataType" , "confidential_http_request" ,
171- "url" , req .GetUrl (),
172- "method" , req .GetMethod (),
173- "bodyKind" , bodyKind ,
174- "bodyLen" , bodyLen ,
175- "headerNames" , slices .Sorted (maps .Keys (req .GetMultiHeaders ())),
176- "templatePublicValueKeys" , slices .Sorted (maps .Keys (req .GetTemplatePublicValues ())),
177- "customRootCACertPEMLen" , len (req .GetCustomRootCaCertPem ()),
178- "timeout" , timeout ,
179- "encryptOutput" , req .GetEncryptOutput ())
180-
181- case types .AppIDConfidentialWorkflows :
182- var execution confworkflowtypes.WorkflowExecution
183- if err := proto .Unmarshal (publicData , & execution ); err != nil {
184- reqLog .Warnw ("failed to decode publicData" ,
185- "event" , "PUBLIC_DATA_DECODE_ERR" ,
186- "appID" , appID ,
187- "publicDataLen" , len (publicData ),
188- "error" , err )
189- return
190- }
191-
192- executeRequestKind := "unset"
193- executeRequestConfigLen := 0
194- var maxResponseSize uint64
195- fields := []any {
196- "event" , "PUBLIC_DATA" ,
197- "appID" , appID ,
198- "publicDataLen" , len (publicData ),
199- "publicDataType" , "workflow_execution" ,
200- "workflowID" , execution .GetWorkflowId (),
201- "executionID" , execution .GetExecutionId (),
202- "owner" , execution .GetOwner (),
203- "orgID" , execution .GetOrgId (),
204- "binaryURL" , execution .GetBinaryUrl (),
205- "binaryHash" , hex .EncodeToString (execution .GetBinaryHash ()),
206- "requirementsPresent" , execution .GetRequirements () != nil ,
207- "restrictionsPresent" , execution .GetRestrictions () != nil ,
208- }
209-
210- if execReq := execution .GetSdkExecuteRequest (); execReq != nil {
211- executeRequestConfigLen = len (execReq .GetConfig ())
212- maxResponseSize = execReq .GetMaxResponseSize ()
213- switch req := execReq .GetRequest ().(type ) {
214- case * sdkpb.ExecuteRequest_Subscribe :
215- executeRequestKind = "subscribe"
216- case * sdkpb.ExecuteRequest_Trigger :
217- executeRequestKind = "trigger"
218- fields = append (fields , "triggerID" , req .Trigger .GetId ())
219- case * sdkpb.ExecuteRequest_PreHook :
220- executeRequestKind = "pre_hook"
221- fields = append (fields , "triggerID" , req .PreHook .GetId ())
222- }
223- }
224-
225- fields = append (fields ,
226- "executeRequestKind" , executeRequestKind ,
227- "executeRequestConfigLen" , executeRequestConfigLen ,
228- "maxResponseSize" , maxResponseSize )
229- reqLog .Infow ("decoded publicData" , fields ... )
230-
231- default :
232- reqLog .Debugw ("publicData decoder unavailable" ,
233- "event" , "PUBLIC_DATA_UNSUPPORTED" ,
234- "appID" , appID ,
235- "publicDataLen" , len (publicData ))
236- }
237- }
238-
239135func NewHostServer (ctx context.Context , clientOverride * http.Client ) * hostServer {
240136 var client * http.Client
241137 if clientOverride == nil {
@@ -261,7 +157,8 @@ func NewHostServer(ctx context.Context, clientOverride *http.Client) *hostServer
261157 config : types.EnclaveConfig {},
262158 verifier : signatureverifier .NewEd25519SignatureVerifier (),
263159 // No-op by default so tests stay quiet; main injects the real logger.
264- logger : cllogger .Sugared (cllogger .Nop ()),
160+ logger : cllogger .Sugared (cllogger .Nop ()),
161+ metrics : noopExecutionMetrics {},
265162 }
266163}
267164
@@ -609,7 +506,7 @@ func (h *hostServer) handleExecute(w http.ResponseWriter, r *http.Request) {
609506 "ephemeralPK" , ephemeralPKHex ,
610507 "bodyLen" , len (body ),
611508 "arrivalTime" , arrivalTime .Format (time .RFC3339Nano ))
612- logPublicData (reqLog , execReq .AppID , execReq .PublicData )
509+ metadata := inspectPublicData (reqLog , execReq .AppID , execReq .PublicData )
613510
614511 // Log hash input components for debugging hash divergence
615512 reqLog .Debugw ("hash inputs" ,
@@ -766,9 +663,16 @@ func (h *hostServer) handleExecute(w http.ResponseWriter, r *http.Request) {
766663 "signers" , signers ,
767664 "signatureCount" , len (requests ))
768665 go func () {
666+ finishExecution := h .metrics .startExecution (metadata )
769667 enclaveStart := time .Now ()
770668 resp , err := h .processBatch (requests )
771669 enclaveDuration := time .Since (enclaveStart )
670+ outcome := executionOutcomeSuccess
671+ if err != nil {
672+ outcome = executionOutcomeError
673+ }
674+ finishExecution (outcome )
675+
772676 if err != nil {
773677 reqLog .Errorw ("enclave execution failed" ,
774678 "event" , "ENCLAVE_ERR" ,
@@ -998,6 +902,9 @@ func main() {
998902 if * writeTimeout <= * quorumTimeout {
999903 log .Fatalf ("write-timeout (%v) must be greater than quorum-timeout (%v)" , * writeTimeout , * quorumTimeout )
1000904 }
905+ if * shutdownTimeout <= 0 {
906+ log .Fatalf ("shutdown-timeout (%v) must be greater than zero" , * shutdownTimeout )
907+ }
1001908
1002909 if * requireBFTQuorum {
1003910 lggr .Infow ("BFT quorum required: batch threshold is 2f+1" , "requireBFTQuorum" , true )
@@ -1012,9 +919,23 @@ func main() {
1012919 sigCh := make (chan os.Signal , 1 )
1013920 signal .Notify (sigCh , syscall .SIGTERM , syscall .SIGINT )
1014921
922+ telemetryCfg := loadHostTelemetryConfig (os .Getenv )
923+ telemetry , err := newHostTelemetry (ctx , telemetryCfg , lggr )
924+ if err != nil {
925+ log .Fatalf ("failed to initialize telemetry: %v" , err )
926+ }
927+ metrics , err := newHostMetrics (telemetry .meter )
928+ if err != nil {
929+ closeCtx , closeCancel := context .WithTimeout (context .Background (), * shutdownTimeout )
930+ _ = telemetry .close (closeCtx )
931+ closeCancel ()
932+ log .Fatalf ("failed to initialize host metrics: %v" , err )
933+ }
934+
1015935 // Start servers. Optionally handle the config endpoint on a different port.
1016936 host := NewHostServer (ctx , nil )
1017937 host .logger = lggr
938+ host .metrics = metrics
1018939 mainMux := http .NewServeMux ()
1019940
1020941 var configServer * http.Server
@@ -1109,7 +1030,7 @@ func main() {
11091030 cancel ()
11101031
11111032 // Give pending requests time to complete
1112- shutdownCtx , shutdownCancel := context .WithTimeout (context .Background (), 30 * time . Second )
1033+ shutdownCtx , shutdownCancel := context .WithTimeout (context .Background (), * shutdownTimeout )
11131034 defer shutdownCancel ()
11141035
11151036 // Shutdown servers gracefully
@@ -1121,6 +1042,9 @@ func main() {
11211042 if err := mainServer .Shutdown (shutdownCtx ); err != nil {
11221043 lggr .Errorw ("main server shutdown error" , "error" , err )
11231044 }
1045+ if err := telemetry .close (shutdownCtx ); err != nil {
1046+ lggr .Errorw ("telemetry shutdown error" , "error" , err )
1047+ }
11241048
11251049 lggr .Infow ("graceful shutdown complete" )
11261050}
0 commit comments