Merge branch 'master' into feature/open-telemetry

Signed-off-by: Dan Rammer <daniel@union.ai>

.github/workflows/master.yml

@@ -35,7 +35,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: '1.17'
+          go-version: '1.18'
       - name: Run GoReleaser
         uses: goreleaser/goreleaser-action@v2
         with:
@@ -54,18 +54,18 @@ jobs:
           fetch-depth: "0"
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.17'
+          go-version: '1.18'
       - name: Unit Tests
         run: make mod_download && make test_unit_codecov
       - name: Push CodeCov
-        uses: codecov/codecov-action@v1
+        uses: codecov/codecov-action@v3.1.1
         with:
           file: coverage.txt
           flags: unittests
-          fail_ci_if_error: true
+          fail_ci_if_error: false
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.17'
+          go-version: '1.18'
       - name: Lint
         run: make install && make lint
       - name: Bench tests

.github/workflows/pull_request.yml

@@ -17,11 +17,11 @@ jobs:
       - name: Unit Tests
         run: make mod_download && make test_unit_codecov
       - name: Push CodeCov
-        uses: codecov/codecov-action@v1.5.2
+        uses: codecov/codecov-action@v3.1.1
         with:
           file: coverage.txt
           flags: unittests
-          fail_ci_if_error: true
+          fail_ci_if_error: false
       - name: Bench tests
         run: make install && make test_benchmark
   lint:

Makefile

@@ -28,4 +28,3 @@ compile:
 .PHONY: test_unit_codecov
 test_unit_codecov:
 	go test ./... -race -coverprofile=coverage.txt -covermode=atomic
-	curl -s https://codecov.io/bash > codecov_bash.sh && bash codecov_bash.sh

contextutils/context.go

@@ -24,6 +24,7 @@ const (
 	RoutineLabelKey    Key = "routine"
 	LaunchPlanIDKey    Key = "lp"
 	ResourceVersionKey Key = "res_ver"
+	SignalIDKey        Key = "signal"
 )
 
 func (k Key) String() string {
@@ -126,6 +127,11 @@ func WithTaskType(ctx context.Context, taskType string) context.Context {
 	return context.WithValue(ctx, TaskTypeKey, taskType)
 }
 
+// Gets a new context with SignalID set.
+func WithSignalID(ctx context.Context, signalID string) context.Context {
+	return context.WithValue(ctx, SignalIDKey, signalID)
+}
+
 // Gets a new context with Go Routine label key set and a label assigned to the context using pprof.Labels.
 // You can then call pprof.SetGoroutineLabels(ctx) to annotate the current go-routine and have that show up in
 // pprof analysis.

contextutils/context_test.go

@@ -102,6 +102,13 @@ func TestWithTaskID(t *testing.T) {
 	assert.Equal(t, "task", ctx.Value(TaskIDKey))
 }
 
+func TestWithSignalID(t *testing.T) {
+	ctx := context.Background()
+	assert.Nil(t, ctx.Value(SignalIDKey))
+	ctx = WithSignalID(ctx, "signal")
+	assert.Equal(t, "signal", ctx.Value(SignalIDKey))
+}
+
 func TestGetFields(t *testing.T) {
 	ctx := context.Background()
 	ctx = WithJobID(WithNamespace(ctx, "ns123"), "job123")

storage/protobuf_store.go

@@ -99,7 +99,11 @@ func newProtoMetrics(scope promutils.Scope) *protoMetrics {
 	}
 }
 
-func NewDefaultProtobufStore(store RawStore, metrics *protoMetrics) DefaultProtobufStore {
+func NewDefaultProtobufStore(store RawStore, scope promutils.Scope) DefaultProtobufStore {
+	return NewDefaultProtobufStoreWithMetrics(store, newProtoMetrics(scope))
+}
+
+func NewDefaultProtobufStoreWithMetrics(store RawStore, metrics *protoMetrics) DefaultProtobufStore {
 	return DefaultProtobufStore{
 		RawStore: store,
 		metrics:  metrics,

storage/protobuf_store_test.go

@@ -157,7 +157,7 @@ func TestDefaultProtobufStore_HardErrors(t *testing.T) {
 			return nil, fmt.Errorf(dummyReadErrorMsg)
 		},
 	}
-	pbErroneousStore := NewDefaultProtobufStore(store, metrics.protoMetrics)
+	pbErroneousStore := NewDefaultProtobufStoreWithMetrics(store, metrics.protoMetrics)
 	t.Run("Test if hard write errors are handled correctly", func(t *testing.T) {
 		err := pbErroneousStore.WriteProtobuf(ctx, k1, Options{}, &mockProtoMessage{X: 5})
 		assert.False(t, IsFailedWriteToCache(err))

storage/rawstores.go

@@ -106,7 +106,7 @@ func (ds *DataStore) RefreshConfig(cfg *Config) error {
 	}
 
 	rawStore = newCachedRawStore(cfg, rawStore, ds.metrics.cacheMetrics)
-	protoStore := NewDefaultProtobufStore(rawStore, ds.metrics.protoMetrics)
+	protoStore := NewDefaultProtobufStoreWithMetrics(rawStore, ds.metrics.protoMetrics)
 	newDS := NewCompositeDataStore(NewURLPathConstructor(), protoStore)
 	newDS.metrics = ds.metrics
 	*ds = *newDS