sync: support for table name prefix

jitsucom · Sep 5, 2023 · 10f966a · 10f966a
1 parent 4a107e4
commit 10f966a
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 19 deletions.
diff --git a/bulkerlib/options.go b/bulkerlib/options.go
@@ -10,7 +10,7 @@ type StreamOption func(*StreamOptions)
 var optionsRegistry = make(map[string]ParseableOption)
 
 // Not used by bulker. Just added here to be treated as known options and don't print errors
-var ignoredOptions = []string{"functions", "streams", "dataLayout", "events", "hosts", "schedule", "timezone", "storageKey", "multithreading"}
+var ignoredOptions = []string{"functions", "streams", "dataLayout", "events", "hosts", "schedule", "timezone", "storageKey", "tableNamePrefix", "multithreading"}
 
 var (
 	BatchSizeOption = ImplementationOption[int]{

diff --git a/sync-controller/job_runner.go b/sync-controller/job_runner.go
@@ -307,8 +307,10 @@ func (j *JobRunner) createPod(podName string, task TaskDescriptor, configuration
 		"PACKAGE":            task.Package,
 		"PACKAGE_VERSION":    task.PackageVersion,
 		"COMMAND":            task.TaskType,
-		"STARTED_AT":         task.StartedAt,
+		"TABLE_NAME_PREFIX":  task.TableNamePrefix,
 		"DATABASE_URL":       databaseURL,
+		"STARTED_BY":         task.StartedBy,
+		"STARTED_AT":         task.StartedAt,
 	}
 	if task.SyncID != "" {
 		sideCarEnv["SYNC_ID"] = task.SyncID

diff --git a/sync-controller/task.go b/sync-controller/task.go
@@ -7,15 +7,17 @@ import (
 )
 
 type TaskDescriptor struct {
-	TaskID         string `json:"taskId"`
-	TaskType       string `json:"taskType"` //spec, discover, read, check
-	SyncID         string `json:"syncId"`
-	SourceType     string `json:"sourceType"`
-	StorageKey     string `json:"storageKey"`
-	Protocol       string `json:"protocol"`
-	Package        string `json:"package"`
-	PackageVersion string `json:"packageVersion"`
-	StartedAt      string `json:"startedAt"`
+	TaskID          string `json:"taskId"`
+	TaskType        string `json:"taskType"` //spec, discover, read, check
+	SyncID          string `json:"syncId"`
+	SourceType      string `json:"sourceType"`
+	StorageKey      string `json:"storageKey"`
+	Protocol        string `json:"protocol"`
+	Package         string `json:"package"`
+	PackageVersion  string `json:"packageVersion"`
+	TableNamePrefix string `json:"tableNamePrefix"`
+	StartedBy       string `json:"startedBy"`
+	StartedAt       string `json:"startedAt"`
 }
 
 func (t *TaskDescriptor) StartedAtTime() time.Time {

diff --git a/sync-controller/task_manager.go b/sync-controller/task_manager.go
@@ -105,12 +105,14 @@ func (t *TaskManager) ReadHandler(c *gin.Context) {
 		taskConfig.State = map[string]any{}
 	}
 	taskDescriptor := TaskDescriptor{
-		TaskType:       "read",
-		Package:        c.Query("package"),
-		PackageVersion: c.Query("version"),
-		SyncID:         c.Query("syncId"),
-		TaskID:         c.Query("taskId"),
-		StartedAt:      time.Now().Format(time.RFC3339),
+		TaskType:        "read",
+		Package:         c.Query("package"),
+		PackageVersion:  c.Query("version"),
+		SyncID:          c.Query("syncId"),
+		TaskID:          c.Query("taskId"),
+		TableNamePrefix: c.Query("tableNamePrefix"),
+		StartedBy:       c.Query("startedBy"),
+		StartedAt:       time.Now().Format(time.RFC3339),
 	}
 
 	taskStatus := t.jobRunner.CreatePod(taskDescriptor, &taskConfig)

diff --git a/sync-sidecar/main.go b/sync-sidecar/main.go
@@ -34,6 +34,7 @@ type AbstractSideCar struct {
 	databaseURL string
 	dbpool      *pgxpool.Pool
 
+	startedBy string
 	startedAt time.Time
 
 	//first error occurred during command
@@ -63,7 +64,7 @@ func main() {
 		startedAt:       startedAt,
 	}
 	if command == "read" {
-		sidecar := &ReadSideCar{AbstractSideCar: abstract}
+		sidecar := &ReadSideCar{AbstractSideCar: abstract, tableNamePrefix: os.Getenv("TABLE_NAME_PREFIX")}
 		sidecar.Run()
 	} else {
 		sidecar := SpecCatalogSideCar{AbstractSideCar: abstract}

diff --git a/sync-sidecar/read.go b/sync-sidecar/read.go
@@ -104,6 +104,7 @@ func (s *ActiveStream) RegisterError(err error) {
 
 type ReadSideCar struct {
 	*AbstractSideCar
+	tableNamePrefix string
 
 	currentStream    *ActiveStream
 	processedStreams map[string]*StreamStat
@@ -299,7 +300,7 @@ func (s *ReadSideCar) openStream(streamName string) *ActiveStream {
 
 	bulkerConnFunc := func(streamReader *io.PipeReader) {
 		s.log("Creating bulker stream: %s mode: %s primary keys: %s", streamName, mode, str.GetPrimaryKeys())
-		bulkerUrl := fmt.Sprintf("%s/bulk/%s?tableName=%s&mode=%s&taskId=%s", s.bulkerURL, s.syncId, url.QueryEscape(streamName), mode, s.taskId)
+		bulkerUrl := fmt.Sprintf("%s/bulk/%s?tableName=%s&mode=%s&taskId=%s", s.bulkerURL, s.syncId, url.QueryEscape(s.tableNamePrefix+streamName), mode, s.taskId)
 		for _, v := range str.GetPrimaryKeys() {
 			bulkerUrl += fmt.Sprintf("&pk=%s", url.QueryEscape(v))
 		}