Fix logging issues in stellar-etl writing to airflow

cmd/command_utils.go

@@ -127,7 +127,7 @@ func maybeUpload(cloudCredentials, cloudStorageBucket, cloudProvider, path strin
 	}
 
 	if len(cloudStorageBucket) == 0 {
-		cmdLogger.Error("No bucket specified")
+		cmdLogger.Fatal("No bucket specified")
 		return
 	}
 
@@ -137,11 +137,11 @@ func maybeUpload(cloudCredentials, cloudStorageBucket, cloudProvider, path strin
 		cloudStorage = newGCS(cloudCredentials, cloudStorageBucket)
 		err := cloudStorage.UploadTo(cloudCredentials, cloudStorageBucket, path)
 		if err != nil {
-			cmdLogger.Errorf("Unable to upload output to GCS: %s", err)
+			cmdLogger.Fatalf("Unable to upload output to GCS: %s", err)
 			return
 		}
 	default:
-		cmdLogger.Error("Unknown cloud provider")
+		cmdLogger.Fatal("Unknown cloud provider")
 	}
 }
 

cmd/export_assets.go

@@ -58,7 +58,7 @@ var assetsCmd = &cobra.Command{
 			seenIDs[transformed.AssetID] = true
 			numBytes, err := exportEntry(transformed, outFile, commonArgs.Extra)
 			if err != nil {
-				cmdLogger.Error(err)
+				cmdLogger.LogError(err)
 				numFailures += 1
 				continue
 			}

cmd/export_effects.go

@@ -1,6 +1,8 @@
 package cmd
 
 import (
+	"fmt"
+
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/stellar/stellar-etl/internal/input"
@@ -34,7 +36,7 @@ var effectsCmd = &cobra.Command{
 			effects, err := transform.TransformEffect(transformInput.Transaction, LedgerSeq, transformInput.LedgerCloseMeta, env.NetworkPassphrase)
 			if err != nil {
 				txIndex := transformInput.Transaction.Index
-				cmdLogger.Errorf("could not transform transaction %d in ledger %d: %v", txIndex, LedgerSeq, err)
+				cmdLogger.LogError(fmt.Errorf("could not transform transaction %d in ledger %d: %v", txIndex, LedgerSeq, err))
 				numFailures += 1
 				continue
 			}

cmd/export_ledger_entry_changes.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/stellar/go/ingest/ledgerbackend"
 	"github.com/stellar/go/xdr"
@@ -28,6 +29,7 @@ confirmed by the Stellar network.
 If no data type flags are set, then by default all of them are exported. If any are set, it is assumed that the others should not
 be exported.`,
 	Run: func(cmd *cobra.Command, args []string) {
+		cmdLogger.SetLevel(logrus.InfoLevel)
 		commonArgs := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = commonArgs.StrictExport
 		env := utils.GetEnvironmentDetails(commonArgs)
@@ -316,7 +318,7 @@ func exportTransformedData(
 				case transform.ClaimableBalanceOutput:
 					// Skipping ClaimableBalanceOutputParquet because it is not needed in the current scope of work
 					// Note that ClaimableBalanceOutputParquet uses nested structs that will need to be handled
-					// for parquet conversio
+					// for parquet conversion
 					skip = true
 				case transform.ConfigSettingOutput:
 					transformedResource = append(transformedResource, v)

internal/input/changes.go

@@ -128,7 +128,10 @@ func extractBatch(
 				if !ok {
 					// TODO: once LedgerEntryTypeData is tracked as well, all types should be addressed,
 					// so this info log should be a warning.
-					logger.Infof("change type: %v not tracked", change.Type)
+					// Skip LedgerEntryTypeData as we are intentionally not processing it
+					if change.Type != xdr.LedgerEntryTypeData {
+						logger.Infof("change type: %v not tracked", change.Type)
+					}
 				} else {
 					cache.AddChange(change)
 				}