<commit_before>package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"gopkg.in/natefinch/lumberjack.v2"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/klog/v2"
)

func main() {
	os.Exit(run())
}

func run() int {
	terminationLog := flag.String("termination-log-file", "", "Write logs after SIGTERM to this file (in addition to stderr)")
	terminationLock := flag.String("termination-touch-file", "", "Touch this file on SIGTERM and delete on termination")
	kubeconfigPath := flag.String("kubeconfig", "", "Optional kubeconfig used to create events")

	klog.InitFlags(nil)
	flag.Set("v", "9")

	// never log to stderr, only through our termination log writer (which sends it also to stderr)
	flag.Set("logtostderr", "false")
	flag.Set("stderrthreshold", "99")

	flag.Parse()
	args := flag.CommandLine.Args()

	if len(args) == 0 {
		fmt.Println("Missing command line")
		return 1
	}

	// use special tee-like writer when termination log is set
	termCh := make(chan struct{})
	var stderr io.Writer = os.Stderr
	var terminationLogger *terminationFileWriter
	if len(*terminationLog) > 0 {
		terminationLogger = &terminationFileWriter{
			Writer:             os.Stderr,
			fn:                 *terminationLog,
			startFileLoggingCh: termCh,
		}
		stderr = terminationLogger

		// do the klog file writer dance: klog writes to all outputs of lower
		// severity. No idea why. So we discard for anything other than info.
		// Otherwise, we would see errors multiple times.
		klog.SetOutput(ioutil.Discard)
		klog.SetOutputBySeverity("INFO", stderr)
	}

	var client kubernetes.Interface
	if len(*kubeconfigPath) > 0 {
		loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{ExplicitPath: *kubeconfigPath}, &clientcmd.ConfigOverrides{})
		if cfg, err := loader.ClientConfig(); err != nil {
			klog.Errorf("failed to load kubeconfig %q: %v", *kubeconfigPath, err)
			return 1
		} else {
			client = kubernetes.NewForConfigOrDie(cfg)
		}
	}

	// touch file early. If the file is not removed on termination, we are not
	// terminating cleanly via SIGTERM.
	if len(*terminationLock) > 0 {
		ref, err := eventReference()
		if err != nil {
			klog.Errorf("failed to get event target: %v", err)
			return 1
		}

		if st, err := os.Stat(*terminationLock); err == nil {
			podName := "unknown"
			if v := os.Getenv("POD_NAME"); len(v) > 0 {
				podName = v // pod name is always the same for static pods
			}
			msg := fmt.Sprintf("Previous pod %s started at %s did not terminate gracefully", podName, st.ModTime().String())

			klog.Warning(msg)
			_, _ = terminationLogger.WriteToTerminationLog([]byte(msg + "\n"))

			if client != nil {
				go wait.PollUntil(5*time.Second, func() (bool, error) {
					if err := eventf(client.CoreV1().Events(ref.Namespace), *ref, corev1.EventTypeWarning, "NonGracefulTermination", msg); err != nil {
						return false, nil
					}

					select {
					case <-termCh:
					default:
					}
					return true, nil
				}, termCh)
			}

			klog.Infof("Deleting old termination lock file %q", *terminationLock)
			if err := os.Remove(*terminationLock); err != nil {
				klog.Errorf("Old termination lock file deletion failed: %v", err)
			}
		}

		// separation to see where the new one is starting
		_, _ = terminationLogger.WriteToTerminationLog([]byte("---\n"))

		klog.Infof("Touching termination lock file %q", *terminationLock)
		if err := touch(*terminationLock); err != nil {
			klog.Infof("Error touching %s: %v", *terminationLock, err)
			// keep going
		}

		defer func() {
			klog.Infof("Deleting termination lock file %q", *terminationLock)
			if err := os.Remove(*terminationLock); err != nil {
				klog.Errorf("Termination lock file deletion failed: %v", err)
			}
		}()
	}

	cmd := exec.Command(args[0], args[1:]...)
	cmd.Stdout = os.Stdout
	cmd.Stderr = stderr

	// forward SIGTERM and SIGINT to child
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for s := range sigCh {
			select {
			case <-termCh:
			default:
				close(termCh)
			}

			klog.Infof("Received signal %s. Forwarding to sub-process %q.", s, args[0])

			cmd.Process.Signal(s)
		}
	}()

	klog.Infof("Launching sub-process %q", cmd)
	rc := 0
	if err := cmd.Run(); err != nil {<commit_msg>UPSTREAM: <carry>: watch-termination: avoid false positives of NonGracefulTermination events<commit_after>package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"os/exec"
	"os/signal"
	"strings"
	"sync"
	"syscall"
	"time"

	"gopkg.in/natefinch/lumberjack.v2"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/kubernetes"
	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/klog/v2"
)

func main() {
	os.Exit(run())
}

func run() int {
	terminationLog := flag.String("termination-log-file", "", "Write logs after SIGTERM to this file (in addition to stderr)")
	terminationLock := flag.String("termination-touch-file", "", "Touch this file on SIGTERM and delete on termination")
	kubeconfigPath := flag.String("kubeconfig", "", "Optional kubeconfig used to create events")
	gracefulTerminatioPeriod := flag.Duration("graceful-termination-duration", 105*time.Second, "The duration of the graceful termination period, e.g. 105s")

	klog.InitFlags(nil)
	flag.Set("v", "9")

	// never log to stderr, only through our termination log writer (which sends it also to stderr)
	flag.Set("logtostderr", "false")
	flag.Set("stderrthreshold", "99")

	flag.Parse()
	args := flag.CommandLine.Args()

	if len(args) == 0 {
		fmt.Println("Missing command line")
		return 1
	}

	// use special tee-like writer when termination log is set
	termCh := make(chan struct{})
	var stderr io.Writer = os.Stderr
	var terminationLogger *terminationFileWriter
	if len(*terminationLog) > 0 {
		terminationLogger = &terminationFileWriter{
			Writer:             os.Stderr,
			fn:                 *terminationLog,
			startFileLoggingCh: termCh,
		}
		stderr = terminationLogger

		// do the klog file writer dance: klog writes to all outputs of lower
		// severity. No idea why. So we discard for anything other than info.
		// Otherwise, we would see errors multiple times.
		klog.SetOutput(ioutil.Discard)
		klog.SetOutputBySeverity("INFO", stderr)
	}

	var client kubernetes.Interface
	if len(*kubeconfigPath) > 0 {
		loader := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(&clientcmd.ClientConfigLoadingRules{ExplicitPath: *kubeconfigPath}, &clientcmd.ConfigOverrides{})
		if cfg, err := loader.ClientConfig(); err != nil {
			klog.Errorf("failed to load kubeconfig %q: %v", *kubeconfigPath, err)
			return 1
		} else {
			client = kubernetes.NewForConfigOrDie(cfg)
		}
	}

	// touch file early. If the file is not removed on termination, we are not
	// terminating cleanly via SIGTERM.
	if len(*terminationLock) > 0 {
		ref, err := eventReference()
		if err != nil {
			klog.Errorf("failed to get event target: %v", err)
			return 1
		}

		if st, err := os.Stat(*terminationLock); err == nil {
			podName := "unknown"
			if v := os.Getenv("POD_NAME"); len(v) > 0 {
				podName = v // pod name is always the same for static pods
			}
			msg := fmt.Sprintf("Previous pod %s started at %s did not terminate gracefully", podName, st.ModTime().String())

			klog.Warning(msg)
			_, _ = terminationLogger.WriteToTerminationLog([]byte(msg + "\n"))

			if client != nil {
				go wait.PollUntil(5*time.Second, func() (bool, error) {
					if err := eventf(client.CoreV1().Events(ref.Namespace), *ref, corev1.EventTypeWarning, "NonGracefulTermination", msg); err != nil {
						return false, nil
					}

					select {
					case <-termCh:
					default:
					}
					return true, nil
				}, termCh)
			}

			klog.Infof("Deleting old termination lock file %q", *terminationLock)
			if err := os.Remove(*terminationLock); err != nil {
				klog.Errorf("Old termination lock file deletion failed: %v", err)
			}
		}

		// separation to see where the new one is starting
		_, _ = terminationLogger.WriteToTerminationLog([]byte("---\n"))

		klog.Infof("Touching termination lock file %q", *terminationLock)
		if err := touch(*terminationLock); err != nil {
			klog.Infof("Error touching %s: %v", *terminationLock, err)
			// keep going
		}

		var deleteLockOnce sync.Once

		if *gracefulTerminatioPeriod > 2*time.Second {
			go func() {
				<-termCh
				<-time.After(*gracefulTerminatioPeriod - 2*time.Second)

				deleteLockOnce.Do(func() {
					klog.Infof("Graceful termination time nearly passed and kube-apiserver has still not terminated. Deleting termination lock file %q to avoid a false positive.", *terminationLock)
					if err := os.Remove(*terminationLock); err != nil {
						klog.Errorf("Termination lock file deletion failed: %v", err)
					}

					if err := eventf(client.CoreV1().Events(ref.Namespace), *ref, corev1.EventTypeWarning, "GracefulTerminationTimeout", "kube-apiserver did not terminate within %s", *gracefulTerminatioPeriod); err != nil {
						klog.Error(err)
					}
				})
			}()
		}

		defer deleteLockOnce.Do(func() {
			klog.Infof("Deleting termination lock file %q", *terminationLock)
			if err := os.Remove(*terminationLock); err != nil {
				klog.Errorf("Termination lock file deletion failed: %v", err)
			}
		})
	}

	cmd := exec.Command(args[0], args[1:]...)
	cmd.Stdout = os.Stdout
	cmd.Stderr = stderr

	// forward SIGTERM and SIGINT to child
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for s := range sigCh {
			select {
			case <-termCh:
			default:
				close(termCh)
			}

			klog.Infof("Received signal %s. Forwarding to sub-process %q.", s, args[0])

			cmd.Process.Signal(s)
		}
	}()

	klog.Infof("Launching sub-process %q", cmd)
	rc := 0
	if err := cmd.Run(); err != nil {
{'fraction_non_alphanumeric': 0.13806879080190979, 'fraction_numerical': 0.0077949917178213, 'mean_word_length': 3.6234234234234233, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 13, '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': '1182406', 'n_tokens_mistral': 4011, 'n_tokens_neox': 3491, 'n_words': 932}