<commit_before>package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"io"
	"net/http"
	"strings"
	"time"

	"github.com/jpillora/backoff"
	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

type tailState int

const (
	tailStateNormal tailState = iota
	tailStateRecover
)

type LogEvent struct {
	Pod       *v1.Pod
	Container *v1.Container
	Timestamp *time.Time
	Message   string
}

type LogEventFunc func(LogEvent)

func NewContainerTailer(
	client kubernetes.Interface,
	pod v1.Pod,
	container v1.Container,
	eventFunc LogEventFunc,
	fromTimestamp *time.Time) *ContainerTailer {
	return &ContainerTailer{
		client:        client,
		pod:           pod,
		container:     container,
		eventFunc:     eventFunc,
		fromTimestamp: fromTimestamp,
		errorBackoff:  &backoff.Backoff{},
		state:         tailStateNormal,
	}
}

type ContainerTailer struct {
	client           kubernetes.Interface
	pod              v1.Pod
	container        v1.Container
	stop             bool
	eventFunc        LogEventFunc
	fromTimestamp    *time.Time
	errorBackoff     *backoff.Backoff
	lastLineChecksum []byte
	state            tailState
}

func (ct *ContainerTailer) Stop() {
	ct.stop = true
}

func (ct *ContainerTailer) Run(ctx context.Context, onError func(err error)) {
	ct.errorBackoff.Reset()
	for !ct.stop {
		stream, err := ct.getStream(ctx)
		if err != nil {
			time.Sleep(ct.errorBackoff.Duration())
			onError(err)
			continue
		}
		if stream == nil {
			break
		}
		if err := ct.runStream(stream); err != nil {
			onError(err)
			time.Sleep(ct.errorBackoff.Duration())
		}
		ct.state = tailStateRecover
	}
}

func (ct *ContainerTailer) runStream(stream io.ReadCloser) error {
	defer func() {
		_ = stream.Close()
	}()

	r := bufio.NewReader(stream)
	for !ct.stop {
		line, err := r.ReadString('\n')
		if err != nil {
			if err != io.EOF {
				return err
			}
			return nil
		}
		ct.errorBackoff.Reset()
		ct.receiveLine(line)
	}
	return nil
}

func (ct *ContainerTailer) receiveLine(s string) {
	if len(s) > 0 && s[len(s)-1] == '\n' {
		s = s[0 : len(s)-1]
	}
	for len(s) > 0 && s[len(s)-1] == '\r' {
		s = s[0 : len(s)-1]
	}

	parts := strings.SplitN(s, " ", 2)
	if len(parts) < 2 {<commit_msg>fix: fixes case where container termination may stop log tailing too early.<commit_after>package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"io"
	"net/http"
	"strings"
	"sync/atomic"
	"time"

	"github.com/jpillora/backoff"
	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
)

type tailState int

const (
	tailStateNormal tailState = iota
	tailStateRecover
)

type LogEvent struct {
	Pod       *v1.Pod
	Container *v1.Container
	Timestamp *time.Time
	Message   string
}

type LogEventFunc func(LogEvent)

func NewContainerTailer(
	client kubernetes.Interface,
	pod v1.Pod,
	container v1.Container,
	eventFunc LogEventFunc,
	fromTimestamp *time.Time) *ContainerTailer {
	return &ContainerTailer{
		client:        client,
		pod:           pod,
		container:     container,
		eventFunc:     eventFunc,
		fromTimestamp: fromTimestamp,
		errorBackoff:  &backoff.Backoff{},
		state:         tailStateNormal,
	}
}

type ContainerTailer struct {
	client           kubernetes.Interface
	pod              v1.Pod
	container        v1.Container
	stop             atomic.Bool
	eventFunc        LogEventFunc
	fromTimestamp    *time.Time
	errorBackoff     *backoff.Backoff
	lastLineChecksum []byte
	state            tailState
}

func (ct *ContainerTailer) Stop() {
	ct.stop.Store(true)
}

func (ct *ContainerTailer) Run(ctx context.Context, onError func(err error)) {
	ct.errorBackoff.Reset()
	for !ct.stop.Load() {
		stream, err := ct.getStream(ctx)
		if err != nil {
			time.Sleep(ct.errorBackoff.Duration())
			onError(err)
			continue
		}
		if stream == nil {
			break
		}
		if err := ct.runStream(stream); err != nil {
			onError(err)
			time.Sleep(ct.errorBackoff.Duration())
		}
		ct.state = tailStateRecover
	}
}

func (ct *ContainerTailer) runStream(stream io.ReadCloser) error {
	defer func() {
		_ = stream.Close()
	}()

	r := bufio.NewReader(stream)
	for {
		line, err := r.ReadString('\n')
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		ct.errorBackoff.Reset()
		ct.receiveLine(line)
	}
	return nil
}

func (ct *ContainerTailer) receiveLine(s string) {
	if len(s) > 0 && s[len(s)-1] == '\n' {
		s = s[0 : len(s)-1]
	}
	for len(s) > 0 && s[len(s)-1] == '\r' {
		s = s[0 : len(s)-1]
	}

	parts := strings.SplitN(s, " ", 2)
	if len(parts) < 2 {
{'fraction_non_alphanumeric': 0.13291818767446853, 'fraction_numerical': 0.011165986686708181, 'mean_word_length': 3.046915725456125, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '3562401', 'n_tokens_mistral': 1905, 'n_tokens_neox': 1787, 'n_words': 379}