Merge branch 'master' into merge/1.33.x

.github/workflows/builds.yml

@@ -107,7 +107,32 @@ jobs:
           password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
+      - name: Build image (no-push)
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: ${{ matrix.build-config.platform }}
+          push: false
+          tags: ${{ matrix.build-config.tags }}
+          labels: ${{ matrix.build-config.labels }}
+          build-args: |
+            BUILD_DATE=${{ needs.docker-oss-meta.outputs.build-date }}
+            VERSION=${{ needs.docker-oss-meta.outputs.version }}
+            COMMIT_HASH=${{ github.sha }}
+            REVISION=${{ needs.docker-oss-meta.outputs.revision }}
+          outputs: type=docker,dest=${{ runner.temp }}/local-rudder-server.tar
+          cache-to: type=local,dest=${{ runner.temp }}/.buildx-cache,mode=max,tag=temp
+      - name: Run TruffleHog
+        run: |
+          docker pull trufflesecurity/trufflehog:latest
+          docker run --rm -v ${{ runner.temp }}:/tmp \
+            trufflesecurity/trufflehog:latest \
+            docker --image file:///tmp/local-rudder-server.tar \
+            --github-actions \
+            --no-verification \
+            --fail
       - name: Build and push
+        if: success()
         uses: docker/build-push-action@v6
         with:
           context: .
@@ -120,7 +145,7 @@ jobs:
             VERSION=${{ needs.docker-oss-meta.outputs.version }}
             COMMIT_HASH=${{ github.sha }}
             REVISION=${{ needs.docker-oss-meta.outputs.revision }}
-
+          cache-from: type=local,src=${{ runner.temp }}/.buildx-cache,tag=temp
   docker-ent-meta:
     runs-on: ubuntu-latest
     outputs:
@@ -181,7 +206,33 @@ jobs:
         with:
           username: rudderlabs
           password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Build image (no-push)
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: ${{ matrix.build-config.platform }}
+          push: false
+          tags: ${{ matrix.build-config.tags }}
+          labels: ${{ matrix.build-config.labels }}
+          build-args: |
+            BUILD_DATE=${{ needs.docker-ent-meta.outputs.build-date }}
+            VERSION=${{ needs.docker-ent-meta.outputs.version }}
+            COMMIT_HASH=${{ github.sha }}
+            REVISION=${{ needs.docker-ent-meta.outputs.revision }}
+            ENTERPRISE_TOKEN=${{ secrets.ENTERPRISE_TOKEN }}
+          outputs: type=docker,dest=${{ runner.temp }}/local-rudder-server.tar
+          cache-to: type=local,dest=${{ runner.temp }}/.buildx-cache,mode=max,tag=temp
+      - name: Run TruffleHog
+        run: |
+          docker pull trufflesecurity/trufflehog:latest
+          docker run --rm -v ${{ runner.temp }}:/tmp \
+            trufflesecurity/trufflehog:latest \
+            docker --image file:///tmp/local-rudder-server.tar \
+            --github-actions \
+            --no-verification \
+            --fail
       - name: Build and push
+        if: success()
         uses: docker/build-push-action@v6
         with:
           context: .
@@ -195,6 +246,7 @@ jobs:
             COMMIT_HASH=${{ github.sha }}
             REVISION=${{ needs.docker-ent-meta.outputs.revision }}
             ENTERPRISE_TOKEN=${{ secrets.ENTERPRISE_TOKEN }}
+          cache-from: type=local,src=${{ runner.temp }}/.buildx-cache,tag=temp
   docker-sbsvc-meta:
     runs-on: ubuntu-latest
     outputs:
@@ -255,7 +307,33 @@ jobs:
         with:
           username: rudderlabs
           password: ${{ secrets.DOCKERHUB_TOKEN }}
+      - name: Build image (no-push)
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: ${{ matrix.build-config.platform }}
+          file: ./suppression-backup-service/Dockerfile
+          push: false
+          tags: ${{ matrix.build-config.tags }}
+          labels: ${{ matrix.build-config.labels }}
+          build-args: |
+            BUILD_DATE=${{ needs.docker-sbsvc-meta.outputs.build-date }}
+            VERSION=${{ needs.docker-sbsvc-meta.outputs.version }}
+            COMMIT_HASH=${{ github.sha }}
+            REVISION=${{ needs.docker-sbsvc-meta.outputs.revision }}
+          outputs: type=docker,dest=${{ runner.temp }}/suppression-backup-service.tar
+          cache-to: type=local,dest=${{ runner.temp }}/.buildx-cache,mode=max,tag=temp
+      - name: Run TruffleHog
+        run: |
+          docker pull trufflesecurity/trufflehog:latest
+          docker run --rm -v ${{ runner.temp }}:/tmp \
+            trufflesecurity/trufflehog:latest \
+            docker --image file:///tmp/suppression-backup-service.tar \
+            --github-actions \
+            --no-verification \
+            --fail
       - name: Build and push
+        if: success()
         uses: docker/build-push-action@v6
         with:
           context: .
@@ -269,6 +347,7 @@ jobs:
             VERSION=${{ needs.docker-sbsvc-meta.outputs.version }}
             COMMIT_HASH=${{ github.sha }}
             REVISION=${{ needs.docker-sbsvc-meta.outputs.revision }}
+          cache-from: type=local,src=${{ runner.temp }}/.buildx-cache,tag=temp
   create-manifest-docker-oss:
     runs-on: ubuntu-latest
     needs: [ docker-oss, docker-oss-meta ]

.github/workflows/verify.yml

@@ -77,5 +77,5 @@ jobs:
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v6
         with:
-          version: v1.57.1
+          version: v1.60.3
           args: -v

Dockerfile

@@ -2,7 +2,7 @@
 # syntax=docker/dockerfile:1
 
 # GO_VERSION is updated automatically to match go.mod, see Makefile
-ARG GO_VERSION=1.22.5
+ARG GO_VERSION=1.23.1
 ARG ALPINE_VERSION=3.20
 FROM golang:${GO_VERSION}-alpine${ALPINE_VERSION} AS builder
 ARG VERSION

Makefile

@@ -6,7 +6,7 @@ TESTFILE=_testok
 MOUNT_PATH=/local
 
 # go tools versions
-GOLANGCI=github.com/golangci/golangci-lint/cmd/golangci-lint@v1.57.1
+GOLANGCI=github.com/golangci/golangci-lint/cmd/golangci-lint@v1.60.3
 gofumpt=mvdan.cc/gofumpt@latest
 govulncheck=golang.org/x/vuln/cmd/govulncheck@latest
 goimports=golang.org/x/tools/cmd/goimports@latest
@@ -96,11 +96,12 @@ install-tools:
 	$(GO) install $(protoc-gen-go)
 	$(GO) install $(protoc-gen-go-grpc)
 	$(GO) install $(gotestsum)
+
 .PHONY: lint
-lint: fmt ## Run linters on all go files
+lint: fmt  ## Run linters on all go files
 	$(GO) run $(GOLANGCI) run -v
-	$(GO) run $(govulncheck) ./...
 	$(GO) run $(actionlint)
+	@$(MAKE) sec
 
 .PHONY: fmt
 fmt: install-tools ## Formats all go files
@@ -126,3 +127,8 @@ generate-openapi-spec: install-tools
 	  -g html2 \
 	  -o ${MOUNT_PATH}/gateway/openapi
 
+.PHONY: sec
+sec: ## Run security checks
+	$(GO) run $(govulncheck) ./...
+	./build/scan_docker.sh
+	./build/scan_docker.sh -f ./suppression-backup-service/Dockerfile

app/apphandlers/embeddedAppHandler.go

@@ -66,9 +66,6 @@ func (a *embeddedApp) Setup() error {
 	if err := rudderCoreDBValidator(); err != nil {
 		return err
 	}
-	if err := rudderCoreWorkSpaceTableSetup(); err != nil {
-		return err
-	}
 	if err := rudderCoreNodeSetup(); err != nil {
 		return err
 	}

app/apphandlers/gatewayAppHandler.go

@@ -42,9 +42,6 @@ func (a *gatewayApp) Setup() error {
 	if err := rudderCoreDBValidator(); err != nil {
 		return err
 	}
-	if err := rudderCoreWorkSpaceTableSetup(); err != nil {
-		return err
-	}
 	a.setupDone = true
 	return nil
 }

app/apphandlers/processorAppHandler.go

@@ -76,9 +76,6 @@ func (a *processorApp) Setup() error {
 	if err := rudderCoreDBValidator(); err != nil {
 		return err
 	}
-	if err := rudderCoreWorkSpaceTableSetup(); err != nil {
-		return err
-	}
 	if err := rudderCoreNodeSetup(); err != nil {
 		return err
 	}

app/apphandlers/setup.go

@@ -52,10 +52,6 @@ func rudderCoreNodeSetup() error {
 	return validators.InitializeNodeMigrations()
 }
 
-func rudderCoreWorkSpaceTableSetup() error {
-	return validators.CheckAndValidateWorkspaceToken()
-}
-
 // NewRsourcesService produces a rsources.JobService through environment configuration (env variables & config file)
 func NewRsourcesService(deploymentType deployment.Type, shouldSetupSharedDB bool) (rsources.JobService, error) {
 	var rsourcesConfig rsources.JobServiceConfig

archiver/worker.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"path"
+	"path/filepath"
 	"time"
 
 	"github.com/google/uuid"
@@ -155,29 +156,35 @@ func (w *worker) uploadJobs(ctx context.Context, jobs []*jobsdb.JobT) (string, e
 	lastJobCreatedAt := jobs[len(jobs)-1].CreatedAt.UTC()
 	workspaceID := jobs[0].WorkspaceId
 
-	gzWriter := fileuploader.NewGzMultiFileWriter()
 	filePath := path.Join(
 		lo.Must(misc.CreateTMPDIR()),
 		"rudder-backups",
 		w.sourceID,
 		fmt.Sprintf("%d_%d_%s_%s.json.gz", firstJobCreatedAt.Unix(), lastJobCreatedAt.Unix(), workspaceID, uuid.NewString()),
 	)
+	if err := os.MkdirAll(filepath.Dir(filePath), os.ModePerm); err != nil {
+		return "", fmt.Errorf("creating gz file %q: mkdir error: %w", filePath, err)
+	}
+	gzWriter, err := misc.CreateGZ(filePath)
+	if err != nil {
+		return "", fmt.Errorf("create gz writer: %w", err)
+	}
+	defer func() { _ = os.Remove(filePath) }()
 
 	for _, job := range jobs {
 		j, err := marshalJob(job)
 		if err != nil {
 			_ = gzWriter.Close()
 			return "", fmt.Errorf("marshal job: %w", err)
 		}
-		if _, err := gzWriter.Write(filePath, append(j, '\n')); err != nil {
+		if _, err := gzWriter.Write(append(j, '\n')); err != nil {
 			_ = gzWriter.Close()
 			return "", fmt.Errorf("write to file: %w", err)
 		}
 	}
 	if err := gzWriter.Close(); err != nil {
 		return "", fmt.Errorf("close writer: %w", err)
 	}
-	defer func() { _ = os.Remove(filePath) }()
 
 	fileUploader, err := w.storageProvider.GetFileManager(w.lifecycle.ctx, workspaceID)
 	if err != nil {

backend-config/backend_config_test.go

@@ -61,6 +61,48 @@ var sampleBackendConfig = ConfigT{
 	},
 }
 
+var sampleConfigWithConnection = ConfigT{
+	WorkspaceID: sampleWorkspaceID,
+	Sources: []SourceT{
+		{
+			ID:       "1",
+			WriteKey: "d",
+			Enabled:  false,
+		}, {
+			ID:       "2",
+			WriteKey: "d2",
+			Enabled:  false,
+			Destinations: []DestinationT{
+				{
+					ID:                 "d1",
+					Name:               "processor Disabled",
+					IsProcessorEnabled: false,
+				}, {
+					ID:                 "d2",
+					Name:               "processor Enabled",
+					IsProcessorEnabled: true,
+				},
+			},
+		},
+	},
+	Connections: map[string]Connection{
+		"1": {
+			SourceID:         "2",
+			DestinationID:    "d1",
+			Enabled:          true,
+			Config:           map[string]interface{}{"key": "value"},
+			ProcessorEnabled: false,
+		},
+		"2": {
+			SourceID:         "2",
+			DestinationID:    "d2",
+			Enabled:          true,
+			Config:           map[string]interface{}{"key2": "value2"},
+			ProcessorEnabled: true,
+		},
+	},
+}
+
 // This configuration is assumed by all gateway tests and, is returned on Subscribe of mocked backend config
 var sampleFilteredSources = ConfigT{
 	Sources: []SourceT{
@@ -264,6 +306,37 @@ func TestConfigUpdate(t *testing.T) {
 		require.Equal(t, (<-chProcess).Data, map[string]ConfigT{workspaces: sampleFilteredSources})
 		require.Equal(t, (<-chBackend).Data, map[string]ConfigT{workspaces: sampleBackendConfig})
 	})
+
+	t.Run("new config with connections", func(t *testing.T) {
+		var (
+			ctrl        = gomock.NewController(t)
+			ctx, cancel = context.WithCancel(context.Background())
+			workspaces  = "foo"
+			cacheStore  = cache.NewMockCache(ctrl)
+		)
+		defer ctrl.Finish()
+		defer cancel()
+
+		wc := NewMockworkspaceConfig(ctrl)
+		wc.EXPECT().Get(gomock.Eq(ctx)).Return(map[string]ConfigT{workspaces: sampleConfigWithConnection}, nil).Times(1)
+
+		var pubSub pubsub.PublishSubscriber
+		bc := &backendConfigImpl{
+			eb:              &pubSub,
+			workspaceConfig: wc,
+			cache:           cacheStore,
+		}
+		bc.curSourceJSON = map[string]ConfigT{workspaces: sampleBackendConfig2}
+
+		chProcess := pubSub.Subscribe(ctx, string(TopicProcessConfig))
+		chBackend := pubSub.Subscribe(ctx, string(TopicBackendConfig))
+
+		bc.configUpdate(ctx)
+		require.True(t, bc.initialized)
+		require.Equal(t, (<-chProcess).Data, map[string]ConfigT{workspaces: sampleFilteredSources})
+		require.Equal(t, (<-chBackend).Data, map[string]ConfigT{workspaces: sampleConfigWithConnection})
+		require.Equal(t, bc.curSourceJSON[workspaces].Connections, sampleConfigWithConnection.Connections)
+	})
 }
 
 func TestFilterProcessorEnabledDestinations(t *testing.T) {

backend-config/types.go

@@ -94,6 +94,15 @@ type ConfigT struct {
 	Settings        Settings                     `json:"settings"`
 	UpdatedAt       time.Time                    `json:"updatedAt"`
 	Credentials     map[string]Credential        `json:"credentials"`
+	Connections     map[string]Connection        `json:"connections"`
+}
+
+type Connection struct {
+	SourceID         string                 `json:"sourceId"`
+	DestinationID    string                 `json:"destinationId"`
+	Enabled          bool                   `json:"enabled"`
+	Config           map[string]interface{} `json:"config"`
+	ProcessorEnabled bool                   `json:"processorEnabled"`
 }
 
 func (c *ConfigT) SourcesMap() map[string]*SourceT {