change: by default, do not exit on a failing file ingestion, but make sure we log the failed status (#154)

Author: iwilltry42

pkg/client/client.go

@@ -23,6 +23,7 @@ type IngestPathsOpts struct {
 	IsDuplicateFuncName  string
 	Prune                bool // Prune deleted files
 	ErrOnUnsupportedFile bool
+	ExitOnFailedFile     bool
 }
 
 type Client interface {

pkg/client/common.go

@@ -149,9 +149,12 @@ func ingestPaths(ctx context.Context, c Client, opts *IngestPathsOpts, datasetID
 					}
 					defer sem.Release(1)
 
-					ingestedFilesCount++
 					slog.Debug("Ingesting file", "path", absPath, "metadata", currentMetadata)
-					return ingestionFunc(sp, currentMetadata.Metadata[filepath.Base(sp)]) // FIXME: metadata
+					err = ingestionFunc(sp, currentMetadata.Metadata[filepath.Base(sp)]) // FIXME: metadata
+					if err == nil {
+						ingestedFilesCount++
+					}
+					return err
 				})
 				return nil
 			})
@@ -178,13 +181,16 @@ func ingestPaths(ctx context.Context, c Client, opts *IngestPathsOpts, datasetID
 				}
 				defer sem.Release(1)
 
-				ingestedFilesCount++
 				var fileMetadata FileMetadata
 				if len(metadataStack) > 0 {
 					currentMetadata := metadataStack[len(metadataStack)-1]
 					fileMetadata = currentMetadata.Metadata[filepath.Base(path)]
 				}
-				return ingestionFunc(path, fileMetadata)
+				err = ingestionFunc(path, fileMetadata)
+				if err == nil {
+					ingestedFilesCount++
+				}
+				return err
 			})
 		}
 

pkg/client/standalone.go

@@ -67,7 +67,11 @@ func (c *StandaloneClient) ListDatasets(ctx context.Context) ([]types.Dataset, e
 }
 
 func (c *StandaloneClient) Ingest(ctx context.Context, datasetID string, name string, data []byte, opts datastore.IngestOpts) ([]string, error) {
-	return c.Datastore.Ingest(ctx, datasetID, name, data, opts)
+	ids, err := c.Datastore.Ingest(ctx, datasetID, name, data, opts)
+	if err != nil {
+		log.FromCtx(ctx).With("status", "failed").With("error", err.Error()).Error("Ingest failed")
+	}
+	return ids, err
 }
 
 func (c *StandaloneClient) IngestPaths(ctx context.Context, datasetID string, opts *IngestPathsOpts, paths ...string) (int, error) {
@@ -111,7 +115,7 @@ func (c *StandaloneClient) IngestPaths(ctx context.Context, datasetID string, op
 			iopts.IngestionFlows = opts.IngestionFlows
 		}
 
-		_, err = c.Ingest(log.ToCtx(ctx, log.FromCtx(ctx).With("filepath", path)), datasetID, filename, file, iopts)
+		_, err = c.Ingest(log.ToCtx(ctx, log.FromCtx(ctx).With("filepath", path).With("absolute_path", iopts.FileMetadata.AbsolutePath)), datasetID, filename, file, iopts)
 
 		if err != nil && !opts.ErrOnUnsupportedFile && errors.Is(err, &documentloader.UnsupportedFileTypeError{}) {
 			err = nil

pkg/cmd/ingest.go

@@ -35,6 +35,7 @@ type ClientIngestOpts struct {
 	NoCreateDataset       bool   `usage:"Do NOT create the dataset if it doesn't exist" default:"true" env:"KNOW_INGEST_NO_CREATE_DATASET"`
 	DeduplicationFuncName string `usage:"Name of the deduplication function to use" name:"dedupe-func" env:"KNOW_INGEST_DEDUPE_FUNC"`
 	ErrOnUnsupportedFile  bool   `usage:"Error on unsupported file types" default:"false" env:"KNOW_INGEST_ERR_ON_UNSUPPORTED_FILE"`
+	ExitOnFailedFile      bool   `usage:"Exit directly on failed file" default:"false" env:"KNOW_INGEST_EXIT_ON_FAILED_FILE"`
 }
 
 func (s *ClientIngest) Customize(cmd *cobra.Command) {
@@ -80,6 +81,7 @@ func (s *ClientIngest) Run(cmd *cobra.Command, args []string) error {
 		IsDuplicateFuncName:  s.DeduplicationFuncName,
 		Prune:                s.Prune,
 		ErrOnUnsupportedFile: s.ErrOnUnsupportedFile,
+		ExitOnFailedFile:     s.ExitOnFailedFile,
 	}
 
 	if s.FlowsFile != "" {
@@ -119,7 +121,7 @@ func (s *ClientIngest) Run(cmd *cobra.Command, args []string) error {
 
 	filesIngested, err := c.IngestPaths(ctx, datasetID, ingestOpts, filePath)
 	if err != nil {
-		return err
+		return fmt.Errorf("ingested %d files but encountered at least one error: %w", filesIngested, err)
 	}
 
 	fmt.Printf("Ingested %d files from %q into dataset %q (took: %s)\n", filesIngested, filePath, datasetID, time.Since(startTime))

pkg/datastore/ingest.go

@@ -191,7 +191,7 @@ func (s *Datastore) Ingest(ctx context.Context, datasetID string, name string, c
 	startTime := time.Now()
 	docIDs, err := s.Vectorstore.AddDocuments(ctx, docs, datasetID)
 	if err != nil {
-		statusLog.With("component", "vectorstore").Error("Failed to add documents", "error", err)
+		statusLog.With("component", "vectorstore").With("status", "failed").With("error", err.Error()).Error("Failed to add documents")
 		return nil, fmt.Errorf("failed to add documents from file %q: %w", opts.FileMetadata.AbsolutePath, err)
 	}
 	statusLog.Debug("Added documents to vectorstore", "duration", time.Since(startTime))