2024 09 version refresh (#77)

Go 1.22->1.23

.github/workflows/go.yml

@@ -19,15 +19,15 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v3
       with:
-        go-version: '1.22'
+        go-version: '1.23'
 
     - name: Build
       run: go build -v ./...
 
     - name: golangci-lint
       uses: golangci/golangci-lint-action@v3
       with:
-        version: v1.58.2
+        version: v1.61.0
         args: --verbose --timeout 10m --fix=false --config=.golangci.yml ./pkg/...
 
     - name: Generate coverage report for coveralls.io

.golangci.yml

@@ -13,7 +13,6 @@ linters:
     - forbidigo
 linters-settings:
   errcheck:
-     ignore : "" # Remove after upgrading golangci-lint 1.58 : https://github.com/golangci/golangci-lint/issues/4733
      severity: error
      confidence: 0.8
      rules:

go.mod

@@ -1,51 +1,50 @@
 module github.com/capillariesio/capillaries
 
-go 1.22
+go 1.23
 
 require (
-	github.com/aws/aws-sdk-go-v2 v1.26.1
-	github.com/aws/aws-sdk-go-v2/config v1.27.11
-	github.com/aws/aws-sdk-go-v2/service/s3 v1.53.1
+	github.com/aws/aws-sdk-go-v2 v1.31.0
+	github.com/aws/aws-sdk-go-v2/config v1.27.36
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.63.0
 	github.com/fraugster/parquet-go v0.12.0
 	github.com/gocql/gocql v1.6.0
 	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/pkg/sftp v1.13.6
-	github.com/rabbitmq/amqp091-go v1.9.0
-	github.com/sethvargo/go-envconfig v1.0.1
-	github.com/shopspring/decimal v1.3.1
-	github.com/stretchr/testify v1.8.4
+	github.com/rabbitmq/amqp091-go v1.10.0
+	github.com/sethvargo/go-envconfig v1.1.0
+	github.com/shopspring/decimal v1.4.0
+	github.com/stretchr/testify v1.9.0
 	go.uber.org/zap v1.27.0
-	golang.org/x/crypto v0.21.0
-	golang.org/x/text v0.14.0
+	golang.org/x/crypto v0.27.0
+	golang.org/x/text v0.18.0
 	gopkg.in/inf.v0 v0.9.1
 )
 
-require github.com/jmespath/go-jmespath v0.4.0 // indirect
+require gopkg.in/yaml.v3 v3.0.1 // indirect
 
 require (
 	github.com/apache/thrift v0.19.0 // indirect
-	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.2 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.11 // indirect
-	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.5 // indirect
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.34 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.14 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.16.15
-	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.5 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.5 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.0 // indirect
-	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.2 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 // indirect
-	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sso v1.20.5 // indirect
-	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 // indirect
-	github.com/aws/smithy-go v1.20.2 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.18 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.18 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.3.18 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.11.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.3.20 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.20 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.17.18 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.23.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.27.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.31.0 // indirect
+	github.com/aws/smithy-go v1.21.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/sys v0.18.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
+	golang.org/x/sys v0.25.0 // indirect
 )

pkg/custom/py_calc/py_calc.donotcover.go

@@ -59,13 +59,13 @@ func (procDef *PyCalcProcessorDef) Run(logger *l.CapiLogger, pCtx *ctx.MessagePr
 	if err != nil {
 		fullErrorInfo, err := procDef.analyseExecError(codeBase, rawOutput, rawErrors, err)
 		if len(fullErrorInfo) > 0 {
-			logger.ErrorCtx(pCtx, fullErrorInfo)
+			logger.ErrorCtx(pCtx, "%s", fullErrorInfo)
 		}
-		return fmt.Errorf("Python interpreter returned an error: %s", err)
+		return fmt.Errorf("python interpreter returned an error: %s", err)
 	}
 	if cmdCtx.Err() == context.DeadlineExceeded {
 		// Timeout occurred, err.Error() is probably: 'signal: killed'
-		return fmt.Errorf("Python calculation timeout %d s expired;", timeout)
+		return fmt.Errorf("python calculation timeout %d s expired;", timeout)
 	}
 	return procDef.analyseExecSuccess(codeBase, rawOutput, rawErrors, procDef.GetFieldRefs(), rsIn, flushVarsArray)
 }

pkg/custom/py_calc/py_calc.go

@@ -175,7 +175,7 @@ func (procDef *PyCalcProcessorDef) Deserialize(raw json.RawMessage, customProcSe
 	// TODO: deserialize other stuff from raw here if needed
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 
 	return nil
@@ -527,7 +527,7 @@ func (procDef *PyCalcProcessorDef) analyseExecSuccess(codeBase string, rawOutput
 
 	if errors.Len() > 0 {
 		// fmt.Println(fmt.Sprintf("%s\nRaw output below:\n%s\nFull codebase below (may be big):\n%s", logText.String(), rawOutput, codeBase.String()))
-		return fmt.Errorf(errors.String())
+		return fmt.Errorf("%s", errors.String())
 	}
 
 	// fmt.Println(fmt.Sprintf("%s\nRaw output below:\n%s", logText.String(), rawOutput))

pkg/custom/tag_and_denormalize/tag_and_denormalize.go

@@ -90,7 +90,7 @@ func (procDef *TagAndDenormalizeProcessorDef) Deserialize(raw json.RawMessage, _
 	// Later on, checkFieldUsageInCustomProcessor() will verify all fields from procDef.UsedInCriteriaFields are valid reader fields
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 	return nil
 }

pkg/exe/daemon/capidaemon.go

@@ -51,13 +51,13 @@ func main() {
 
 	envConfig, err := env.ReadEnvConfigFile(initCtx, "capidaemon.json")
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatalf("%s", err.Error())
 	}
 	envConfig.CustomProcessorDefFactoryInstance = &StandardDaemonProcessorDefFactory{}
 
 	logger, err := l.NewLoggerFromEnvConfig(envConfig)
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatalf("%s", err.Error())
 	}
 	defer logger.Close()
 

pkg/exe/daemon/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.22 as build
+FROM golang:1.23 as build
 WORKDIR /usr/src/capillaries
 ADD go.mod go.sum ./
 ADD ./pkg ./pkg

pkg/exe/webapi/capiwebapi.go

@@ -662,14 +662,14 @@ func main() {
 
 	envConfig, err := env.ReadEnvConfigFile(initCtx, "capiwebapi.json")
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatalf("%s", err.Error())
 	}
 
 	// Webapi (like toolbelt and daemon) requires custom proc def factory, otherwise it will not be able to start runs
 	envConfig.CustomProcessorDefFactoryInstance = &StandardWebapiProcessorDefFactory{}
 	logger, err := l.NewLoggerFromEnvConfig(envConfig)
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatalf("%s", err.Error())
 	}
 	defer logger.Close()
 
@@ -697,6 +697,6 @@ func main() {
 
 	logger.Info("listening on %d...", h.Env.Webapi.Port)
 	if err := http.ListenAndServe(fmt.Sprintf(":%d", h.Env.Webapi.Port), mux); err != nil {
-		log.Fatalf(err.Error())
+		log.Fatalf("%s", err.Error())
 	}
 }

pkg/exe/webapi/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.22 as golang
+FROM golang:1.23 as golang
 WORKDIR /usr/src/capillaries
 ADD go.mod go.sum ./
 ADD ./pkg ./pkg

pkg/proc/data_util.go

@@ -144,7 +144,7 @@ func selectBatchPagedAllRowids(logger *l.CapiLogger,
 
 	dbWarnings := iter.Warnings()
 	if len(dbWarnings) > 0 {
-		logger.WarnCtx(pCtx, strings.Join(dbWarnings, ";"))
+		logger.WarnCtx(pCtx, "%s", strings.Join(dbWarnings, ";"))
 	}
 
 	rs.RowCount = 0
@@ -196,7 +196,7 @@ func selectBatchFromIdxTablePaged(logger *l.CapiLogger,
 
 	dbWarnings := iter.Warnings()
 	if len(dbWarnings) > 0 {
-		logger.WarnCtx(pCtx, strings.Join(dbWarnings, ";"))
+		logger.WarnCtx(pCtx, "%s", strings.Join(dbWarnings, ";"))
 	}
 
 	rs.RowCount = 0
@@ -247,7 +247,7 @@ func selectBatchFromTableByToken(logger *l.CapiLogger,
 
 	dbWarnings := iter.Warnings()
 	if len(dbWarnings) > 0 {
-		logger.WarnCtx(pCtx, strings.Join(dbWarnings, ";"))
+		logger.WarnCtx(pCtx, "%s", strings.Join(dbWarnings, ";"))
 	}
 	rs.RowCount = 0
 	var lastRetrievedToken int64

pkg/proc/file_inserter.go

@@ -60,7 +60,7 @@ func (instr *FileInserter) checkWorkerOutputForErrors() error {
 			}
 		default:
 			if len(errors) > 0 {
-				return fmt.Errorf(strings.Join(errors, "; "))
+				return fmt.Errorf("%s", strings.Join(errors, "; "))
 			} else {
 				return nil
 			}
@@ -102,7 +102,7 @@ func (instr *FileInserter) waitForWorker(logger *l.CapiLogger, pCtx *ctx.Message
 	logger.DebugCtx(pCtx, "closed BatchesIn")
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 
 	return nil

pkg/proc/file_inserter_csv.go

@@ -77,7 +77,10 @@ func (instr *FileInserter) csvFileInserterWorker(logger *l.CapiLogger) {
 	var errOpen error
 	f, err := os.OpenFile(localFilePath, os.O_APPEND|os.O_WRONLY, 0644)
 	if err != nil {
-		errOpen = fmt.Errorf("cannot open %s(temp %s) for appending: [%s]", instr.FinalFileUrl, instr.TempFilePath, err.Error())
+		errOpen = fmt.Errorf("cannot open csv %s(temp %s) for appending: [%s]", instr.FinalFileUrl, instr.TempFilePath, err.Error())
+	}
+	if f == nil {
+		errOpen = fmt.Errorf("cannot open csv %s(temp %s) for appending: unknown error", instr.FinalFileUrl, instr.TempFilePath)
 	} else {
 		defer f.Close()
 	}

pkg/proc/file_inserter_parquet.go

@@ -107,9 +107,13 @@ func (instr *FileInserter) parquetFileInserterWorker(logger *l.CapiLogger, codec
 	var w *storage.ParquetWriter
 	f, err := os.OpenFile(localFilePath, os.O_APPEND|os.O_WRONLY, 0644)
 	if err != nil {
-		errOpen = fmt.Errorf("cannot open %s(temp %s) for appending: [%s]", instr.FinalFileUrl, instr.TempFilePath, err.Error())
+		errOpen = fmt.Errorf("cannot open parquet %s(temp %s) for appending: [%s]", instr.FinalFileUrl, instr.TempFilePath, err.Error())
 	} else {
-		defer f.Close()
+		if f == nil {
+			errOpen = fmt.Errorf("cannot open parquet %s(temp %s) for appending: unknown error", instr.FinalFileUrl, instr.TempFilePath)
+		} else {
+			defer f.Close()
+		}
 
 		w, err = storage.NewParquetWriter(f, codec)
 		if err != nil {

pkg/proc/table_inserter.go

@@ -154,7 +154,7 @@ func (instr *TableInserter) letWorkersDrainRecordWrittenStatuses(logger *l.CapiL
 	logger.DebugCtx(pCtx, "done draining %d records at RecordsSent=%d from instr.RecordWrittenStatuses, %d errors", drainedRecordCount, instr.RecordsSent, errCount)
 
 	if len(errorsFound) > 0 {
-		return fmt.Errorf(strings.Join(errorsFound, "; "))
+		return fmt.Errorf("%s", strings.Join(errorsFound, "; "))
 	}
 
 	return nil
@@ -167,7 +167,7 @@ func (instr *TableInserter) letWorkersDrainRecordWrittenStatusesAndCloseInserter
 	// If anything was sent at all - drain
 	if instr.RecordsSent > 0 {
 		if err := instr.letWorkersDrainRecordWrittenStatuses(logger, pCtx); err != nil {
-			logger.ErrorCtx(pCtx, fmt.Sprintf("error(s) while waiting for workers to drain RecordsIn: %s", err.Error()))
+			logger.ErrorCtx(pCtx, "error(s) while waiting for workers to drain RecordsIn: %s", err.Error())
 		}
 	}
 
@@ -405,7 +405,7 @@ func (instr *TableInserter) insertDataRecord(logger *l.CapiLogger, writeItem *Wr
 
 		if !errors.Is(err, ErrDuplicateRowid) {
 			errorToReturn := fmt.Errorf("cannot insert data record [%s]: %s", pq.Query, err.Error())
-			logger.ErrorCtx(instr.PCtx, errorToReturn.Error())
+			logger.ErrorCtx(instr.PCtx, "%s", errorToReturn.Error())
 			return curRowid, errorToReturn
 		} else if retryCount < maxRetries-1 {
 			rowidRand.Seed(newSeed(instr.MachineHash))
@@ -415,7 +415,7 @@ func (instr *TableInserter) insertDataRecord(logger *l.CapiLogger, writeItem *Wr
 	}
 
 	errorToReturn := fmt.Errorf("duplicate rowid not written [%s], giving up after %d rowid retries", pq.Query, maxRetries)
-	logger.ErrorCtx(instr.PCtx, errorToReturn.Error())
+	logger.ErrorCtx(instr.PCtx, "%s", errorToReturn.Error())
 	return curRowid, errorToReturn
 }
 
@@ -610,7 +610,7 @@ func (instr *TableInserter) insertDistinctIdxAndDataRecords(logger *l.CapiLogger
 	} // retry loop
 
 	errorToReport := fmt.Errorf("cannot insert distinct idx/data records after %d attempts", maxRetries)
-	logger.ErrorCtx(pCtx, errorToReport.Error())
+	logger.ErrorCtx(pCtx, "%s", errorToReport.Error())
 	return curRowid, errorToReport
 }
 

pkg/sc/field_ref.go

@@ -251,7 +251,7 @@ func checkAllowed(fieldRefsToCheck *FieldRefs, prohibitedFieldRefs *FieldRefs, a
 	}
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 
 	return nil

pkg/sc/file_creator_def.go

@@ -184,7 +184,7 @@ func (creatorDef *FileCreatorDef) CalculateFileRecordFromSrcVars(srcVars eval.Va
 	}
 
 	if len(errors) > 0 {
-		return nil, fmt.Errorf(strings.Join(errors, "; "))
+		return nil, fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 	return fileRecord, nil
 }

pkg/sc/file_reader_def.go

@@ -141,7 +141,7 @@ func (frDef *FileReaderDef) Deserialize(rawReader json.RawMessage) error {
 	}
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 	return nil
 }

pkg/sc/key.go

@@ -154,7 +154,7 @@ func BuildKey(fieldMap map[string]any, idxDef *IdxDef) (string, error) {
 			}
 
 		default:
-			return "", fmt.Errorf(fmt.Sprintf("cannot build key, unsupported field data type %s", comp.FieldType))
+			return "", fmt.Errorf("cannot build key, unsupported field data type %s", comp.FieldType)
 		}
 
 		// Used by file creator top. Not used by actual indexes - Cassandra cannot do proper ORDER BY anyways

pkg/sc/script_def.go

@@ -115,7 +115,7 @@ func (scriptDef *ScriptDef) Deserialize(jsonOrYamlBytesScript []byte, scriptType
 	}
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, "; "))
+		return fmt.Errorf("%s", strings.Join(errors, "; "))
 	}
 
 	// Table -> node map, to look for ord and lkp indexes, for those nodes that create tables

pkg/sc/script_def_loader_test.go

@@ -212,7 +212,7 @@ func (procDef *SomeTestCustomProcessorDef) Deserialize(raw json.RawMessage, _ js
 	}
 
 	if len(errors) > 0 {
-		return fmt.Errorf(strings.Join(errors, ";"))
+		return fmt.Errorf("%s", strings.Join(errors, ";"))
 	}
 
 	procDef.UsedInTargetExpressionsFields = GetFieldRefsUsedInAllTargetExpressions(procDef.ProducedFields)