Skip to content

Commit 5144841

Browse files
ktropsKatie Atrops
ktrops
and
Katie Atrops
authored
IWF-1174: add logging for S3 & local activity errors (#599)
Co-authored-by: Katie Atrops <katiea@indeed.com>
1 parent ed9f8c2 commit 5144841

4 files changed

Lines changed: 186 additions & 14 deletions

File tree

‎config/config.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -111,6 +111,9 @@ type (
111111
InterpreterActivityConfig InterpreterActivityConfig `yaml:"interpreterActivityConfig"`
112112
VerboseDebug bool
113113
FailAtMemoIncompatibility bool
114+
// LogLocalActivityThresholdBytes enables warn-level logging of local activity inputs/outputs when the
115+
// serialized payload size meets or exceeds this value. Set to 0 (default) to disable.
116+
LogLocalActivityThresholdBytes int `yaml:"logLocalActivityThresholdBytes"`
114117
}
115118

116119
TemporalConfig struct {

‎service/common/blobstore/store_impl.go‎

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,23 @@ func (b *blobStoreImpl) WriteObject(ctx context.Context, workflowId, data string
9090
err = putObject(ctx, b.s3Client, b.activeStorage.S3Bucket, b.pathPrefix+path, data)
9191
if err != nil {
9292
b.writeObjectErrorCounter.Inc(1)
93+
var re s3.ResponseError
94+
if errors.As(err, &re) {
95+
b.logger.Error("PutObject S3 API error",
96+
tag.Key("requestId"), tag.Value(re.ServiceRequestID()),
97+
tag.Key("hostId"), tag.Value(re.ServiceHostID()),
98+
tag.Key("bucket"), tag.Value(b.activeStorage.S3Bucket),
99+
tag.Key("workflowId"), tag.Value(workflowId),
100+
tag.Error(err))
101+
err = fmt.Errorf("failed to write object (requestId=%s, hostId=%s): %w",
102+
re.ServiceRequestID(), re.ServiceHostID(), err)
103+
} else {
104+
b.logger.Error("PutObject error",
105+
tag.Key("bucket"), tag.Value(b.activeStorage.S3Bucket),
106+
tag.Key("workflowId"), tag.Value(workflowId),
107+
tag.Error(err))
108+
err = fmt.Errorf("failed to write object: %w", err)
109+
}
93110
return
94111
}
95112
b.writeObjectSuccessHistogram.Record(time.Duration(len(data)))
@@ -105,7 +122,24 @@ func (b *blobStoreImpl) ReadObject(ctx context.Context, storeId, path string) (s
105122
data, err := getObject(ctx, b.s3Client, storeConfig.S3Bucket, b.pathPrefix+path)
106123
if err != nil {
107124
b.readObjectErrorCounter.Inc(1)
108-
return "", err
125+
var re s3.ResponseError
126+
if errors.As(err, &re) {
127+
b.logger.Error("GetObject S3 API error",
128+
tag.Key("requestId"), tag.Value(re.ServiceRequestID()),
129+
tag.Key("hostId"), tag.Value(re.ServiceHostID()),
130+
tag.Key("bucket"), tag.Value(storeConfig.S3Bucket),
131+
tag.Key("path"), tag.Value(path),
132+
tag.Key("storeId"), tag.Value(storeId),
133+
tag.Error(err))
134+
return "", fmt.Errorf("failed to read object (requestId=%s, hostId=%s): %w",
135+
re.ServiceRequestID(), re.ServiceHostID(), err)
136+
}
137+
b.logger.Error("GetObject error",
138+
tag.Key("bucket"), tag.Value(storeConfig.S3Bucket),
139+
tag.Key("path"), tag.Value(path),
140+
tag.Key("storeId"), tag.Value(storeId),
141+
tag.Error(err))
142+
return "", fmt.Errorf("failed to read object: %w", err)
109143
}
110144
b.readObjectSuccessHistogram.Record(time.Duration(len(data)))
111145
return data, nil

‎service/common/blobstore/store_impl_test.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,5 +275,10 @@ func TestBlobStoreIntegration(t *testing.T) {
275275
err = blobStore.DeleteWorkflowObjects(ctx, "invalid-store-id", "some-workflow-path")
276276
assert.Error(t, err)
277277
assert.Contains(t, err.Error(), "store not found")
278+
279+
// Test reading a non-existent key from a valid store triggers the new error wrapping
280+
_, err = blobStore.ReadObject(ctx, testStorageId, "nonexistent/path/that/does/not/exist")
281+
assert.Error(t, err)
282+
assert.Contains(t, err.Error(), "failed to read object")
278283
})
279284
}

‎service/interpreter/activityImpl.go‎

Lines changed: 143 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package interpreter
22

33
import (
44
"context"
5+
"encoding/json"
56
"fmt"
67
"io"
78
"net/http"
@@ -39,9 +40,9 @@ func StateApiWaitUntil(
3940
logger.Info("StateWaitUntilActivity", "input", log.ToJsonAndTruncateForLogging(input))
4041
iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)
4142

42-
svcCfg := env.GetSharedConfig()
43+
sharedCfg := env.GetSharedConfig()
4344
apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
44-
DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
45+
DefaultHeader: sharedCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
4546
Servers: []iwfidl.ServerConfiguration{
4647
{
4748
URL: iwfWorkerBaseUrl,
@@ -59,13 +60,31 @@ func StateApiWaitUntil(
5960
if input.Request.StateInput != nil && input.Request.StateInput.ExtStoreId != nil {
6061
_, err = loadStateInputFromExternalStorage(ctx, input.Request.StateInput)
6162
if err != nil {
63+
if activityInfo.IsLocalActivity {
64+
reqBytes, _ := json.Marshal(input.Request)
65+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
66+
logger.Warn("StateApiWaitUntil local activity return on error",
67+
"workflowId", activityInfo.WorkflowExecution.ID,
68+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
69+
"payloadSize", len(reqBytes))
70+
}
71+
}
6272
return nil, err
6373
}
6474
}
6575

6676
// Load data attributes from external storage
6777
err = blobstore.LoadDataObjectsFromExternalStorage(ctx, input.Request.DataObjects, env.GetBlobStore())
6878
if err != nil {
79+
if activityInfo.IsLocalActivity {
80+
reqBytes, _ := json.Marshal(input.Request)
81+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
82+
logger.Warn("StateApiWaitUntil local activity return on error",
83+
"workflowId", activityInfo.WorkflowExecution.ID,
84+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
85+
"payloadSize", len(reqBytes))
86+
}
87+
}
6988
return nil, err
7089
}
7190

@@ -93,6 +112,15 @@ func StateApiWaitUntil(
93112
Details: &errDetails,
94113
},
95114
})
115+
if activityInfo.IsLocalActivity {
116+
reqBytes, _ := json.Marshal(input.Request)
117+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
118+
logger.Warn("StateApiWaitUntil local activity return on error",
119+
"workflowId", activityInfo.WorkflowExecution.ID,
120+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
121+
"payloadSize", len(reqBytes))
122+
}
123+
}
96124
return nil, stateStartErr
97125
}
98126

@@ -115,6 +143,15 @@ func StateApiWaitUntil(
115143
Details: &errDetails,
116144
},
117145
})
146+
if activityInfo.IsLocalActivity {
147+
reqBytes, _ := json.Marshal(input.Request)
148+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
149+
logger.Warn("StateApiWaitUntil local activity return on error",
150+
"workflowId", activityInfo.WorkflowExecution.ID,
151+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
152+
"payloadSize", len(reqBytes))
153+
}
154+
}
118155
return nil, stateStartErr
119156
}
120157

@@ -125,9 +162,18 @@ func StateApiWaitUntil(
125162
resp.LocalActivityInput = composeInputForDebug(input.Request.Context.GetStateExecutionId())
126163
}
127164

128-
if env.GetSharedConfig().ExternalStorage.Enabled {
165+
if env.GetSharedConfig().ExternalStorage.Enabled && env.GetBlobStore() != nil {
129166
err = blobstore.WriteDataObjectsToExternalStorage(ctx, resp.UpsertDataObjects, activityInfo.WorkflowExecution.ID, env.GetSharedConfig().ExternalStorage.ThresholdInBytes, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage.Enabled)
130167
if err != nil {
168+
if activityInfo.IsLocalActivity {
169+
reqBytes, _ := json.Marshal(input.Request)
170+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
171+
logger.Warn("StateApiWaitUntil local activity return on error",
172+
"workflowId", activityInfo.WorkflowExecution.ID,
173+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
174+
"payloadSize", len(reqBytes))
175+
}
176+
}
131177
return nil, err
132178
}
133179
}
@@ -143,6 +189,15 @@ func StateApiWaitUntil(
143189
EndTimestampInMs: ptr.Any(time.Now().UnixMilli()),
144190
SearchAttributes: searchAttributes,
145191
})
192+
if activityInfo.IsLocalActivity {
193+
respBytes, _ := json.Marshal(resp)
194+
if threshold := sharedCfg.Interpreter.LogLocalActivityThresholdBytes; threshold > 0 && len(respBytes) >= threshold {
195+
logger.Warn("StateApiWaitUntil local activity return on success",
196+
"workflowId", activityInfo.WorkflowExecution.ID,
197+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
198+
"payloadSize", len(respBytes))
199+
}
200+
}
146201
return resp, nil
147202
}
148203

@@ -166,9 +221,9 @@ func StateApiExecute(
166221
logger.Info("StateExecuteActivity", "input", log.ToJsonAndTruncateForLogging(input))
167222

168223
iwfWorkerBaseUrl := urlautofix.FixWorkerUrl(input.IwfWorkerUrl)
169-
svcCfg := env.GetSharedConfig()
224+
sharedCfg := env.GetSharedConfig()
170225
apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
171-
DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
226+
DefaultHeader: sharedCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
172227
Servers: []iwfidl.ServerConfiguration{
173228
{
174229
URL: iwfWorkerBaseUrl,
@@ -187,13 +242,31 @@ func StateApiExecute(
187242
if input.Request.StateInput != nil && input.Request.StateInput.ExtStoreId != nil {
188243
wholeStateInputCopy, err = loadStateInputFromExternalStorage(ctx, input.Request.StateInput)
189244
if err != nil {
245+
if activityInfo.IsLocalActivity {
246+
reqBytes, _ := json.Marshal(input.Request)
247+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
248+
logger.Warn("StateApiExecute local activity return on error",
249+
"workflowId", activityInfo.WorkflowExecution.ID,
250+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
251+
"payloadSize", len(reqBytes))
252+
}
253+
}
190254
return nil, err
191255
}
192256
}
193257

194258
// Load data attributes from external storage
195259
err = blobstore.LoadDataObjectsFromExternalStorage(ctx, input.Request.DataObjects, env.GetBlobStore())
196260
if err != nil {
261+
if activityInfo.IsLocalActivity {
262+
reqBytes, _ := json.Marshal(input.Request)
263+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
264+
logger.Warn("StateApiExecute local activity return on error",
265+
"workflowId", activityInfo.WorkflowExecution.ID,
266+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
267+
"payloadSize", len(reqBytes))
268+
}
269+
}
197270
return nil, err
198271
}
199272

@@ -233,6 +306,15 @@ func StateApiExecute(
233306
Details: &errDetails,
234307
},
235308
})
309+
if activityInfo.IsLocalActivity {
310+
reqBytes, _ := json.Marshal(input.Request)
311+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
312+
logger.Warn("StateApiExecute local activity return on error",
313+
"workflowId", activityInfo.WorkflowExecution.ID,
314+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
315+
"payloadSize", len(reqBytes))
316+
}
317+
}
236318
return nil, stateApiExecuteErr
237319
}
238320

@@ -255,6 +337,15 @@ func StateApiExecute(
255337
Details: &errDetails,
256338
},
257339
})
340+
if activityInfo.IsLocalActivity {
341+
reqBytes, _ := json.Marshal(input.Request)
342+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
343+
logger.Warn("StateApiExecute local activity return on error",
344+
"workflowId", activityInfo.WorkflowExecution.ID,
345+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
346+
"payloadSize", len(reqBytes))
347+
}
348+
}
258349
return nil, stateApiExecuteErr
259350
}
260351

@@ -265,13 +356,32 @@ func StateApiExecute(
265356
resp.LocalActivityInput = composeInputForDebug(input.Request.Context.GetStateExecutionId())
266357
}
267358

268-
if env.GetSharedConfig().ExternalStorage.Enabled {
359+
// Externalize only when enabled and blob store is available (nil when e.g. STAGING_LEVEL was empty at worker start).
360+
if env.GetSharedConfig().ExternalStorage.Enabled && env.GetBlobStore() != nil {
269361
resp.StateDecision.NextStates, err = writeNextStateInputsToExternalStorage(ctx, resp.StateDecision.NextStates, wholeStateInputCopy, activityInfo.WorkflowExecution.ID)
270362
if err != nil {
363+
if activityInfo.IsLocalActivity {
364+
reqBytes, _ := json.Marshal(input.Request)
365+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
366+
logger.Warn("StateApiExecute local activity return on error",
367+
"workflowId", activityInfo.WorkflowExecution.ID,
368+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
369+
"payloadSize", len(reqBytes))
370+
}
371+
}
271372
return nil, err
272373
}
273374
err = blobstore.WriteDataObjectsToExternalStorage(ctx, resp.UpsertDataObjects, activityInfo.WorkflowExecution.ID, env.GetSharedConfig().ExternalStorage.ThresholdInBytes, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage.Enabled)
274375
if err != nil {
376+
if activityInfo.IsLocalActivity {
377+
reqBytes, _ := json.Marshal(input.Request)
378+
if sharedCfg.Interpreter.LogLocalActivityThresholdBytes > 0 {
379+
logger.Warn("StateApiExecute local activity return on error",
380+
"workflowId", activityInfo.WorkflowExecution.ID,
381+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
382+
"payloadSize", len(reqBytes))
383+
}
384+
}
275385
return nil, err
276386
}
277387
}
@@ -287,6 +397,15 @@ func StateApiExecute(
287397
EndTimestampInMs: ptr.Any(time.Now().UnixMilli()),
288398
SearchAttributes: input.Request.SearchAttributes,
289399
})
400+
if activityInfo.IsLocalActivity {
401+
respBytes, _ := json.Marshal(resp)
402+
if threshold := sharedCfg.Interpreter.LogLocalActivityThresholdBytes; threshold > 0 && len(respBytes) >= threshold {
403+
logger.Warn("StateApiExecute local activity return on success",
404+
"workflowId", activityInfo.WorkflowExecution.ID,
405+
"stateExecutionId", input.Request.Context.GetStateExecutionId(),
406+
"payloadSize", len(respBytes))
407+
}
408+
}
290409
return resp, nil
291410
}
292411

@@ -438,11 +557,11 @@ func DumpWorkflowInternal(
438557
logger := provider.GetLogger(ctx)
439558
logger.Info("DumpWorkflowInternalActivity", "input", log.ToJsonAndTruncateForLogging(req))
440559

441-
svcCfg := env.GetSharedConfig()
442-
apiAddress := svcCfg.GetApiServiceAddressWithDefault()
560+
sharedCfg := env.GetSharedConfig()
561+
apiAddress := sharedCfg.GetApiServiceAddressWithDefault()
443562

444563
apiClient := iwfidl.NewAPIClient(&iwfidl.Configuration{
445-
DefaultHeader: svcCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
564+
DefaultHeader: sharedCfg.Interpreter.InterpreterActivityConfig.DefaultHeaders,
446565
Servers: []iwfidl.ServerConfiguration{
447566
{
448567
URL: apiAddress,
@@ -466,14 +585,25 @@ func InvokeWorkerRpc(
466585
provider := interfaces.GetActivityProviderByType(backendType)
467586
logger := provider.GetLogger(ctx)
468587
logger.Info("InvokeWorkerRpcActivity", "input", log.ToJsonAndTruncateForLogging(req))
588+
activityInfo := provider.GetActivityInfo(ctx)
589+
sharedCfg := env.GetSharedConfig()
469590

470-
apiMaxSeconds := env.GetSharedConfig().Api.MaxWaitSeconds
591+
apiMaxSeconds := sharedCfg.Api.MaxWaitSeconds
471592

472-
resp, statusErr := rpc.InvokeWorkerRpc(ctx, rpcPrep, req, apiMaxSeconds, env.GetBlobStore(), env.GetSharedConfig().ExternalStorage)
473-
return &interfaces.InvokeRpcActivityOutput{
593+
resp, statusErr := rpc.InvokeWorkerRpc(ctx, rpcPrep, req, apiMaxSeconds, env.GetBlobStore(), sharedCfg.ExternalStorage)
594+
output := &interfaces.InvokeRpcActivityOutput{
474595
RpcOutput: resp,
475596
StatusError: statusErr,
476-
}, nil
597+
}
598+
if activityInfo.IsLocalActivity {
599+
outputBytes, _ := json.Marshal(output)
600+
if threshold := sharedCfg.Interpreter.LogLocalActivityThresholdBytes; threshold > 0 && len(outputBytes) >= threshold {
601+
logger.Warn("InvokeWorkerRpc local activity return",
602+
"workflowId", activityInfo.WorkflowExecution.ID,
603+
"payloadSize", len(outputBytes))
604+
}
605+
}
606+
return output, nil
477607
}
478608

479609
func writeNextStateInputsToExternalStorage(ctx context.Context, nextStates []iwfidl.StateMovement, currentInputCopy *iwfidl.EncodedObject, workflowId string) ([]iwfidl.StateMovement, error) {

0 commit comments

Comments
 (0)