[feat] - add cancellation to s3 source

[ahrav]

Description:

This PR handles cancellation during a s3 scan.

Checklist:

• Tests passing (make test-community)?
• Lint passing (make lint this requires golangci-lint)?

[mcastorina]

It looks like most of the cancellation happens via contexts, which means the workers we want to cancel should already be context aware.

Can this be done with a standard context.WithCancel?

[ahrav]

It looks like most of the cancellation happens via contexts, which means the workers we want to cancel should already be context aware.

Can this be done with a standard context.WithCancel?

Would you mind providing an example? I couldn't get it to work :( I was having a problem getting the error out of pageChunker

[ahrav]

It looks like most of the cancellation happens via contexts, which means the workers we want to cancel should already be context aware.

Can this be done with a standard context.WithCancel?

Would you mind providing an example? I couldn't get it to work :( I was having a problem getting the error out of pageChunker

If you can get it to work feel free to close this. I'll be out next week, so feel free to replace this PR :)

Just gotta make sure the resumption test in #3570 works.

[mcastorina]

I'll be out most of next week too, but for posterity another avenue could be errgroup.WithContext.

[ahrav]

I'll be out most of next week too, but for posterity another avenue could be errgroup.WithContext.

hmm.. I guess we could do something like:

        // Start a goroutine to watch for errors from pageChunker
        go func() {
            select {
            case err := <-errChan:
                if err != nil {
                    cancel() // Cancel the context if we get an error
                }
            case <-scanCtx.Done():
                return
            }
        }()

        input := &s3.ListObjectsV2Input{Bucket: &bucket}
        if bucket == resumePoint.CurrentBucket && resumePoint.StartAfter != "" {
            input.StartAfter = &resumePoint.StartAfter
        }

        err = regionalClient.ListObjectsV2PagesWithContext(
            scanCtx, // Use the cancellable context
            input,
            func(page *s3.ListObjectsV2Output, _ bool) bool {
                pageMetadata := pageMetadata{
                    bucket:     bucket,
                    pageNumber: pageNumber,
                    client:     regionalClient,
                    page:       page,
                }
                processingState := processingState{
                    errorCount:  &errorCount,
                    objectCount: &objectCount,
                }

                // Process the page and send any errors to errChan
                if err := s.pageChunker(scanCtx, pageMetadata, processingState, chunksChan); err != nil {
                    select {
                    case errChan <- err:
                    default:
                    }
                    return false
                }

                if scanCtx.Err() != nil {
                    return false
                }

                pageNumber++
                return true
            })

        // Check if we stopped due to context cancellation
        if scanCtx.Err() != nil {
            return scanCtx.Err()
        }

        if err != nil {
            if role == "" {
                bucketCtx.Logger().Error(err, "could not list objects in bucket")
            } else {
                bucketCtx.Logger().V(3).Info("could not list objects in bucket", "err", err)
            }
        }
    }

Idk why, but that feels more complicated 😅 (assuming, this is what you meant?)