<commit_before>package processor

import (
	"context"
	"fmt"
	"log"
	"sync/atomic"
	"time"

	"github.com/atlassian/smith"
	"github.com/atlassian/smith/pkg/processor/graph"
	"github.com/atlassian/smith/pkg/resources"

	"github.com/cenk/backoff"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	unstructured_conversion "k8s.io/apimachinery/pkg/conversion/unstructured"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
)

var (
	converter = unstructured_conversion.NewConverter(false)
)

type workerConfig struct {
	bundleClient *rest.RESTClient
	scDynamic    dynamic.ClientPool
	clients      dynamic.ClientPool
	rc           ReadyChecker
	deepCopy     smith.DeepCopy
	store        smith.ByNameStore
}

type worker struct {
	bundleRef
	workerConfig
	bo             backoff.BackOff
	workRequests   chan<- workRequest
	notifyRequests chan<- notifyRequest

	// --- Items below are only modified from main processor goroutine ---
	pendingBundle *smith.Bundle   // next work item storage
	notify        chan<- struct{} // channel to close to notify sleeping worker about work available
	// ---

	needsRebuild int32 // must be accessed via atomics only
}

func (wrk *worker) rebuildLoop(ctx context.Context, done func()) {
	defer done()
	var timer *time.Timer
	defer func() {
		if timer != nil {
			timer.Stop()
		}
	}()
	work := make(chan *smith.Bundle)
	workReq := workRequest{
		bundleRef: wrk.bundleRef,
		work:      work,
	}
	var notifyChan <-chan struct{}
	var sleepChan <-chan time.Time
	requestWork := true
	for {
		if requestWork {
			select {
			case <-ctx.Done():
				return
			case wrk.workRequests <- workReq:
				// New work has been requested
			}
		} else {
			requestWork = true
		}
		select {
		case <-ctx.Done():
			return
		case <-notifyChan:
			// Woke up because work is available
			sleepChan = nil
			notifyChan = nil
			timer.Stop()
			timer = nil
		case <-sleepChan:
			// Woke up after backoff sleep
			sleepChan = nil
			notifyChan = nil
			timer = nil
		case bundle, ok := <-work:
			if !ok {
				// No work available, exit
				return
			}
			// Work available
			isReady, retriable, err := wrk.rebuild(bundle)
			if wrk.handleRebuildResult(bundle, isReady, retriable, err) {
				// There was a retriable error
				next := wrk.bo.NextBackOff()
				if next == backoff.Stop {
					wrk.bo.Reset() // reset the backoff to restart it
					next = wrk.bo.NextBackOff()
				}
				timer = time.NewTimer(next)
				sleepChan = timer.C
				nc := make(chan struct{})
				notifyChan = nc
				nrq := notifyRequest{
					bundleRef: wrk.bundleRef,
					bundle:    bundle,
					notify:    nc,
				}
				select {
				case <-ctx.Done():
					return
				case wrk.notifyRequests <- nrq:
					// Asked processor to notify us if/when new work is available
				}
				requestWork = false
			}
		}
	}
}

// Parse bundle, build resource graph, traverse graph, assert each resource exists.
// For each resource ensure its dependencies (if any) are in READY state before creating it.
// If at least one dependency is not READY, exit loop. Rebuild will/should be called once the dependency
// updates it's state (noticed via watching).

// READY state might mean something different for each resource type. For ThirdPartyResources it may mean
// that a field "State" in the Status of the resource is set to "Ready". It may be customizable via
// annotations with some defaults.
func (wrk *worker) rebuild(bundle *smith.Bundle) (isReady, retriableError bool, e error) {
	log.Printf("[WORKER][%s/%s] Rebuilding bundle", wrk.namespace, wrk.bundleName)

	// Build resource map by name
	resourceMap := make(map[smith.ResourceName]smith.Resource, len(bundle.Spec.Resources))
	for _, res := range bundle.Spec.Resources {
		if _, exist := resourceMap[res.Name]; exist {
			return false, false, fmt.Errorf("bundle %s/%s contains two resources with the same name %q", wrk.namespace, wrk.bundleName, res.Name)
		}
		resourceMap[res.Name] = res
	}

	// Build the graph and topologically sort it
	graphData, sortErr := graph.TopologicalSort(bundle)
	if sortErr != nil {
		return false, false, sortErr
	}

	readyResources := make(map[smith.ResourceName]*unstructured.Unstructured, len(bundle.Spec.Resources))
	allReady := true

	// Visit vertices in sorted order
nextVertex:
	for _, v := range graphData.SortedVertices {
		if wrk.checkNeedsRebuild() {
			return false, false, nil
		}

		// Check if all resource dependencies are ready (so we can start processing this one)
		for _, dependency := range graphData.Graph.Vertices[v].Edges() {
			if _, ok := readyResources[dependency]; !ok {
				allReady = false
				log.Printf("[WORKER][%s/%s] Dependencies are not ready for resource %q", wrk.namespace, wrk.bundleName, v)
				continue nextVertex // Move to the next resource
			}
		}
		// Process the resource
		log.Printf("[WORKER][%s/%s] Checking resource %q", wrk.namespace, wrk.bundleName, v)
		res := resourceMap[v]
		readyResource, retriable, err := wrk.checkResource(bundle, &res, readyResources)
		if err != nil {
			return false, retriable, err
		}
		log.Printf("[WORKER][%s/%s] Resource %q, ready: %t", wrk.namespace, wrk.bundleName, v, readyResource != nil)
		if readyResource != nil {
			readyResources[v] = readyResource
		} else {
			allReady = false
		}
	}
	return allReady, false, nil
}<commit_msg>Improve logging<commit_after>package processor

import (
	"context"
	"fmt"
	"log"
	"sync/atomic"
	"time"

	"github.com/atlassian/smith"
	"github.com/atlassian/smith/pkg/processor/graph"
	"github.com/atlassian/smith/pkg/resources"

	"github.com/cenk/backoff"
	"k8s.io/apimachinery/pkg/api/equality"
	"k8s.io/apimachinery/pkg/api/errors"
	kerrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	unstructured_conversion "k8s.io/apimachinery/pkg/conversion/unstructured"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
)

var (
	converter = unstructured_conversion.NewConverter(false)
)

type workerConfig struct {
	bundleClient *rest.RESTClient
	scDynamic    dynamic.ClientPool
	clients      dynamic.ClientPool
	rc           ReadyChecker
	deepCopy     smith.DeepCopy
	store        smith.ByNameStore
}

type worker struct {
	bundleRef
	workerConfig
	bo             backoff.BackOff
	workRequests   chan<- workRequest
	notifyRequests chan<- notifyRequest

	// --- Items below are only modified from main processor goroutine ---
	pendingBundle *smith.Bundle   // next work item storage
	notify        chan<- struct{} // channel to close to notify sleeping worker about work available
	// ---

	needsRebuild int32 // must be accessed via atomics only
}

func (wrk *worker) rebuildLoop(ctx context.Context, done func()) {
	defer done()
	var timer *time.Timer
	defer func() {
		if timer != nil {
			timer.Stop()
		}
	}()
	work := make(chan *smith.Bundle)
	workReq := workRequest{
		bundleRef: wrk.bundleRef,
		work:      work,
	}
	var notifyChan <-chan struct{}
	var sleepChan <-chan time.Time
	requestWork := true
	for {
		if requestWork {
			select {
			case <-ctx.Done():
				return
			case wrk.workRequests <- workReq:
				// New work has been requested
			}
		} else {
			requestWork = true
		}
		select {
		case <-ctx.Done():
			return
		case <-notifyChan:
			// Woke up because work is available
			sleepChan = nil
			notifyChan = nil
			timer.Stop()
			timer = nil
		case <-sleepChan:
			// Woke up after backoff sleep
			sleepChan = nil
			notifyChan = nil
			timer = nil
		case bundle, ok := <-work:
			if !ok {
				// No work available, exit
				return
			}
			// Work available
			isReady, retriable, err := wrk.rebuild(bundle)
			if wrk.handleRebuildResult(bundle, isReady, retriable, err) {
				// There was a retriable error
				next := wrk.bo.NextBackOff()
				if next == backoff.Stop {
					wrk.bo.Reset() // reset the backoff to restart it
					next = wrk.bo.NextBackOff()
				}
				timer = time.NewTimer(next)
				sleepChan = timer.C
				nc := make(chan struct{})
				notifyChan = nc
				nrq := notifyRequest{
					bundleRef: wrk.bundleRef,
					bundle:    bundle,
					notify:    nc,
				}
				select {
				case <-ctx.Done():
					return
				case wrk.notifyRequests <- nrq:
					// Asked processor to notify us if/when new work is available
				}
				requestWork = false
			}
		}
	}
}

// Parse bundle, build resource graph, traverse graph, assert each resource exists.
// For each resource ensure its dependencies (if any) are in READY state before creating it.
// If at least one dependency is not READY, exit loop. Rebuild will/should be called once the dependency
// updates it's state (noticed via watching).

// READY state might mean something different for each resource type. For ThirdPartyResources it may mean
// that a field "State" in the Status of the resource is set to "Ready". It may be customizable via
// annotations with some defaults.
func (wrk *worker) rebuild(bundle *smith.Bundle) (isReady, retriableError bool, e error) {
	log.Printf("[WORKER][%s/%s] Rebuilding bundle", wrk.namespace, wrk.bundleName)

	// Build resource map by name
	resourceMap := make(map[smith.ResourceName]smith.Resource, len(bundle.Spec.Resources))
	for _, res := range bundle.Spec.Resources {
		if _, exist := resourceMap[res.Name]; exist {
			return false, false, fmt.Errorf("bundle contains two resources with the same name %q", res.Name)
		}
		resourceMap[res.Name] = res
	}

	// Build the graph and topologically sort it
	graphData, sortErr := graph.TopologicalSort(bundle)
	if sortErr != nil {
		return false, false, sortErr
	}

	readyResources := make(map[smith.ResourceName]*unstructured.Unstructured, len(bundle.Spec.Resources))
	allReady := true

	// Visit vertices in sorted order
nextVertex:
	for _, v := range graphData.SortedVertices {
		if wrk.checkNeedsRebuild() {
			return false, false, nil
		}

		// Check if all resource dependencies are ready (so we can start processing this one)
		for _, dependency := range graphData.Graph.Vertices[v].Edges() {
			if _, ok := readyResources[dependency]; !ok {
				allReady = false
				log.Printf("[WORKER][%s/%s] Dependency %q is required by resource %q but it's not ready", wrk.namespace, wrk.bundleName, dependency, v)
				continue nextVertex // Move to the next resource
			}
		}
		// Process the resource
		log.Printf("[WORKER][%s/%s] Checking resource %q", wrk.namespace, wrk.bundleName, v)
		res := resourceMap[v]
		readyResource, retriable, err := wrk.checkResource(bundle, &res, readyResources)
		if err != nil {
			return false, retriable, err
		}
		log.Printf("[WORKER][%s/%s] Resource %q, ready: %t", wrk.namespace, wrk.bundleName, v, readyResource != nil)
		if readyResource != nil {
			readyResources[v] = readyResource
		} else {
			allReady = false
		}
	}
	return allReady, false, nil
}
{'fraction_non_alphanumeric': 0.11277224067921743, 'fraction_numerical': 0.0018456995201181247, 'mean_word_length': 3.264856355765447, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5796089', 'n_tokens_mistral': 3946, 'n_tokens_neox': 3478, 'n_words': 1033}