[CSE-1] Improve logging of CSV export (#285)

* [CSE-1] Improve logging of CSV export

- Standardised the fields coming out of the regular exporter
- Added timing information from the CSV exporter

* Encode ts in JSON logs as ISO timestamps

internal/app/feed/exporter_csv.go

@@ -22,7 +22,10 @@ type CSVExporter struct {
 
 // CreateSchema generated schema for a feed in csv format
 func (e *CSVExporter) CreateSchema(feed Feed, rows interface{}) error {
-	e.Logger.Infof("%s: writing out CSV schema file", feed.Name())
+	logger := e.Logger.With(
+		"feed", feed.Name(),
+	)
+	logger.Info("writing out CSV schema file")
 
 	exportFilePath := filepath.Join(e.ExportPath, fmt.Sprintf("%s.csv", feed.Name()))
 	_, err := os.Stat(exportFilePath)
@@ -36,14 +39,17 @@ func (e *CSVExporter) CreateSchema(feed Feed, rows interface{}) error {
 		return gocsv.Marshal(rows, file)
 	}
 
-	e.Logger.Infof("%s: skipping. CSV file already exists.", feed.Name())
+	logger.Info("CSV file already exists, skipping")
 
 	return nil
 }
 
 // FinaliseExport closes out an export
 func (e *CSVExporter) FinaliseExport(feed Feed, rows interface{}) error {
-	e.Logger.Infof("%s: writing out CSV file", feed.Name())
+	logger := e.Logger.With(
+		"feed", feed.Name(),
+	)
+	logger.Info("writing out CSV file")
 
 	err := e.cleanOldFiles(feed.Name())
 	if err != nil {
@@ -58,6 +64,7 @@ func (e *CSVExporter) FinaliseExport(feed Feed, rows interface{}) error {
 	rowsAdded := 0
 	var file *os.File
 	for {
+		preTime := time.Now()
 		resp := e.DB.Table(feed.Name()).
 			Order(feed.Order()).
 			Limit(limit).
@@ -66,6 +73,7 @@ func (e *CSVExporter) FinaliseExport(feed Feed, rows interface{}) error {
 		if resp.Error != nil {
 			return resp.Error
 		}
+		postQueryTime := time.Now()
 
 		if resp.RowsAffected == 0 || resp.RowsAffected == -1 {
 			break
@@ -97,9 +105,20 @@ func (e *CSVExporter) FinaliseExport(feed Feed, rows interface{}) error {
 			}
 		}
 
+		postWriteTime := time.Now()
+
 		offset = offset + limit
 		rowsAdded += int(resp.RowsAffected)
+
+		logger.With(
+			"rows_added", resp.RowsAffected,
+			"total_time_ms", postWriteTime.Sub(preTime).Milliseconds(),
+			"query_time_ms", postQueryTime.Sub(preTime).Milliseconds(),
+			"write_time_ms", postWriteTime.Sub(postQueryTime).Milliseconds(),
+		).Info("writing out CSV batch")
 	}
+
+	logger.Info("CSV exported")
 	return nil
 }
 

internal/app/feed/exporter_schema.go

@@ -35,7 +35,7 @@ func isPrimaryKey(pk int) string {
 
 // WriteSchema writes schema of a feed to output in tabular format
 func (e *SchemaExporter) WriteSchema(feed Feed) error {
-	e.Logger.Infof("Schema for %s:", feed.Name())
+	e.Logger.With("feed", feed.Name()).Info("writing schema")
 
 	schema := &[]*schema{}
 

internal/app/feed/exporter_sql.go

@@ -63,7 +63,9 @@ func (e *SQLExporter) InitFeed(feed Feed, opts *InitFeedOptions) error {
 	}
 
 	if opts.Truncate {
-		e.Logger.Infof("%s: truncating", feed.Name())
+		e.Logger.With(
+			"feed", feed.Name(),
+		).Info("truncating")
 		result := e.DB.Session(&gorm.Session{AllowGlobalUpdate: true}).Unscoped().Delete(model)
 		if result.Error != nil {
 			return errors.Wrap(result.Error, "Unable to truncate table")

internal/app/feed/feed.go

@@ -2,7 +2,6 @@ package feed
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	"github.com/SafetyCulture/iauditor-exporter/internal/app/api"
@@ -44,33 +43,3 @@ type Exporter interface {
 	SupportsUpsert() bool
 	ParameterLimit() int
 }
-
-// LogStringConfig is the config for GetLogString function
-type LogStringConfig struct {
-	RemainingRecords int64
-	HTTPDuration     time.Duration
-	ExporterDuration time.Duration
-}
-
-// GetLogString build a log string based on input arguments
-func GetLogString(feedName string, cfg *LogStringConfig) string {
-	var args = []any{feedName}
-	var format = "%s: "
-
-	if cfg != nil {
-		format = format + "%d remaining."
-		args = append(args, cfg.RemainingRecords)
-
-		if cfg.HTTPDuration.Milliseconds() != 0 {
-			format = format + " Last http call was %dms."
-			args = append(args, cfg.HTTPDuration.Milliseconds())
-		}
-
-		if cfg.ExporterDuration.Milliseconds() != 0 {
-			format = format + " Last export operation was %dms."
-			args = append(args, cfg.ExporterDuration.Milliseconds())
-		}
-	}
-
-	return fmt.Sprintf(format, args...)
-}

internal/app/feed/feed_action.go

@@ -95,8 +95,10 @@ func (f *ActionFeed) CreateSchema(exporter Exporter) error {
 
 // Export exports the feed to the supplied exporter
 func (f *ActionFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Delete data if incremental refresh is disabled so there is no duplicates
@@ -107,7 +109,9 @@ func (f *ActionFeed) Export(ctx context.Context, apiClient *api.Client, exporter
 	f.ModifiedAfter, err = exporter.LastModifiedAt(f, f.ModifiedAfter, orgID)
 	util.Check(err, "unable to load modified after")
 
-	logger.Infof("%s: exporting for org_id: %s since: %s", feedName, orgID, f.ModifiedAfter.Format(time.RFC1123))
+	logger.With(
+		"modified_after", f.ModifiedAfter.Format(time.RFC1123),
+	).Info("exporting")
 
 	err = apiClient.DrainFeed(ctx, &api.GetFeedRequest{
 		InitialURL: "/feed/actions",
@@ -136,11 +140,11 @@ func (f *ActionFeed) Export(ctx context.Context, apiClient *api.Client, exporter
 			}
 		}
 
-		logger.Info(GetLogString(f.Name(), &LogStringConfig{
-			RemainingRecords: resp.Metadata.RemainingRecords,
-			HTTPDuration:     apiClient.Duration,
-			ExporterDuration: exporter.GetDuration(),
-		}))
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	})
 	util.Check(err, "Failed to export feed")

internal/app/feed/feed_action_assignees.go

@@ -97,8 +97,10 @@ func (f *ActionAssigneeFeed) writeRows(ctx context.Context, exporter Exporter, r
 
 // Export exports the feed to the supplied exporter
 func (f *ActionAssigneeFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Delete data if incremental refresh is disabled so there is no duplicates
@@ -109,7 +111,9 @@ func (f *ActionAssigneeFeed) Export(ctx context.Context, apiClient *api.Client,
 	f.ModifiedAfter, err = exporter.LastModifiedAt(f, f.ModifiedAfter, orgID)
 	util.Check(err, "unable to load modified after")
 
-	logger.Infof("%s: exporting for org_id: %s since: %s", feedName, orgID, f.ModifiedAfter.Format(time.RFC1123))
+	logger.With(
+		"modified_after", f.ModifiedAfter.Format(time.RFC1123),
+	).Info("exporting")
 
 	err = apiClient.DrainFeed(ctx, &api.GetFeedRequest{
 		InitialURL: "/feed/action_assignees",
@@ -127,7 +131,11 @@ func (f *ActionAssigneeFeed) Export(ctx context.Context, apiClient *api.Client,
 			util.Check(err, "Failed to write data to exporter")
 		}
 
-		logger.Infof("%s: %d remaining", feedName, resp.Metadata.RemainingRecords)
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	})
 	util.Check(err, "Failed to export feed")

internal/app/feed/feed_group.go

@@ -61,10 +61,12 @@ func (f *GroupFeed) CreateSchema(exporter Exporter) error {
 
 // Export exports the feed to the supplied exporter
 func (f *GroupFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
-	logger.Infof("%s: exporting for org_id: %s", feedName, orgID)
+	logger.Info("exporting")
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Truncate files if upserts aren't supported.
@@ -96,7 +98,11 @@ func (f *GroupFeed) Export(ctx context.Context, apiClient *api.Client, exporter
 			}
 		}
 
-		logger.Infof("%s: %d remaining. Last call was %dms", feedName, resp.Metadata.RemainingRecords, apiClient.Duration.Milliseconds())
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	})
 	util.Check(err, "Failed to export feed")

internal/app/feed/feed_group_user.go

@@ -62,10 +62,12 @@ func (f *GroupUserFeed) CreateSchema(exporter Exporter) error {
 
 // Export exports the feed to the supplied exporter
 func (f *GroupUserFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
-	logger.Infof("%s: exporting for org_id: %s", feedName, orgID)
+	logger.Info("exporting")
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Truncate files if upserts aren't supported.
@@ -101,7 +103,11 @@ func (f *GroupUserFeed) Export(ctx context.Context, apiClient *api.Client, expor
 			}
 		}
 
-		logger.Infof("%s: %d remaining. Last call was %dms", feedName, resp.Metadata.RemainingRecords, apiClient.Duration.Milliseconds())
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	})
 	util.Check(err, "Failed to export feed")

internal/app/feed/feed_inspection.go

@@ -161,8 +161,10 @@ func (f *InspectionFeed) CreateSchema(exporter Exporter) error {
 
 // Export exports the feed to the supplied exporter
 func (f *InspectionFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Delete data if incremental refresh is disabled so there is no duplicates
@@ -173,7 +175,10 @@ func (f *InspectionFeed) Export(ctx context.Context, apiClient *api.Client, expo
 	f.ModifiedAfter, err = exporter.LastModifiedAt(f, f.ModifiedAfter, orgID)
 	util.Check(err, "unable to load modified after")
 
-	logger.Infof("%s: exporting for org_id: %s since: %s - %s", feedName, orgID, f.ModifiedAfter.Format(time.RFC1123), f.WebReportLink)
+	logger.With(
+		"modified_after", f.ModifiedAfter.Format(time.RFC1123),
+		"web_report_link_type", f.WebReportLink,
+	).Info("exporting")
 
 	// Process Inspections
 	err = f.processNewInspections(ctx, apiClient, exporter)
@@ -214,11 +219,12 @@ func (f *InspectionFeed) processNewInspections(ctx context.Context, apiClient *a
 			}
 		}
 
-		lg.Info(GetLogString(f.Name(), &LogStringConfig{
-			RemainingRecords: resp.Metadata.RemainingRecords,
-			HTTPDuration:     apiClient.Duration,
-			ExporterDuration: exporter.GetDuration(),
-		}))
+		lg.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
+
 		return nil
 	}
 

internal/app/feed/feed_inspection_item.go

@@ -222,8 +222,10 @@ func (f *InspectionItemFeed) CreateSchema(exporter Exporter) error {
 
 // Export exports the feed to the supplied exporter
 func (f *InspectionItemFeed) Export(ctx context.Context, apiClient *api.Client, exporter Exporter, orgID string) error {
-	logger := util.GetLogger()
-	feedName := f.Name()
+	logger := util.GetLogger().With(
+		"feed", f.Name(),
+		"org_id", orgID,
+	)
 
 	exporter.InitFeed(f, &InitFeedOptions{
 		// Delete data if incremental refresh is disabled so there is no duplicates
@@ -234,7 +236,9 @@ func (f *InspectionItemFeed) Export(ctx context.Context, apiClient *api.Client,
 	f.ModifiedAfter, err = exporter.LastModifiedAt(f, f.ModifiedAfter, orgID)
 	util.Check(err, "unable to load modified after")
 
-	logger.Infof("%s: exporting for org_id: %s since: %s", feedName, orgID, f.ModifiedAfter.Format(time.RFC1123))
+	logger.With(
+		"modified_after", f.ModifiedAfter.Format(time.RFC1123),
+	).Info("exporting")
 
 	err = apiClient.DrainFeed(ctx, &api.GetFeedRequest{
 		InitialURL: "/feed/inspection_items",
@@ -257,11 +261,11 @@ func (f *InspectionItemFeed) Export(ctx context.Context, apiClient *api.Client,
 			util.Check(err, "Failed to write data to exporter")
 		}
 
-		logger.Info(GetLogString(f.Name(), &LogStringConfig{
-			RemainingRecords: resp.Metadata.RemainingRecords,
-			HTTPDuration:     apiClient.Duration,
-			ExporterDuration: exporter.GetDuration(),
-		}))
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	})
 	util.Check(err, "Failed to export feed")

internal/app/feed/feed_issue.go

@@ -3,9 +3,10 @@ package feed
 import (
 	"context"
 	"encoding/json"
+	"time"
+
 	"github.com/SafetyCulture/iauditor-exporter/internal/app/api"
 	"github.com/SafetyCulture/iauditor-exporter/internal/app/util"
-	"time"
 )
 
 const (
@@ -118,7 +119,11 @@ func (f *IssueFeed) Export(ctx context.Context, apiClient *api.Client, exporter
 			}
 		}
 
-		logger.Infof("%s: %d remaining. Last call was %dms", f.Name(), resp.Metadata.RemainingRecords, apiClient.Duration.Milliseconds())
+		logger.With(
+			"estimated_remaining", resp.Metadata.RemainingRecords,
+			"duration_ms", apiClient.Duration.Milliseconds(),
+			"export_duration_ms", exporter.GetDuration().Milliseconds(),
+		).Info("export batch complete")
 		return nil
 	}