<commit_before>package app

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"os/signal"
	"path/filepath"
	"runtime"
	"sync"
	"syscall"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/cobra"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/klog/v2"

	"sigs.k8s.io/apiserver-network-proxy/cmd/server/app/options"
	"sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client"
	"sigs.k8s.io/apiserver-network-proxy/pkg/server"
	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
)

var udsListenerLock sync.Mutex

func NewProxyCommand(p *Proxy, o *options.ProxyRunOptions) *cobra.Command {
	cmd := &cobra.Command{
		Use:  "proxy",
		Long: `A gRPC proxy server, receives requests from the API server and forwards to the agent.`,
		RunE: func(cmd *cobra.Command, args []string) error {
			return p.run(o)
		},
	}

	return cmd
}

type Proxy struct {
}

type StopFunc func()

func (p *Proxy) run(o *options.ProxyRunOptions) error {
	o.Print()
	if err := o.Validate(); err != nil {
		return fmt.Errorf("failed to validate server options with %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var k8sClient *kubernetes.Clientset
	if o.AgentNamespace != "" {
		config, err := clientcmd.BuildConfigFromFlags("", o.KubeconfigPath)
		if err != nil {
			return fmt.Errorf("failed to load kubernetes client config: %v", err)
		}

		if o.KubeconfigQPS != 0 {
			klog.V(1).Infof("Setting k8s client QPS: %v", o.KubeconfigQPS)
			config.QPS = o.KubeconfigQPS
		}
		if o.KubeconfigBurst != 0 {
			klog.V(1).Infof("Setting k8s client Burst: %v", o.KubeconfigBurst)
			config.Burst = o.KubeconfigBurst
		}
		k8sClient, err = kubernetes.NewForConfig(config)
		if err != nil {
			return fmt.Errorf("failed to create kubernetes clientset: %v", err)
		}
	}

	authOpt := &server.AgentTokenAuthenticationOptions{
		Enabled:                o.AgentNamespace != "",
		AgentNamespace:         o.AgentNamespace,
		AgentServiceAccount:    o.AgentServiceAccount,
		KubernetesClient:       k8sClient,
		AuthenticationAudience: o.AuthenticationAudience,
	}
	klog.V(1).Infoln("Starting frontend server for client connections.")
	ps, err := server.GenProxyStrategiesFromStr(o.ProxyStrategies)
	if err != nil {
		return err
	}
	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt, o.WarnOnChannelLimit)

	frontendStop, err := p.runFrontendServer(ctx, o, server)
	if err != nil {
		return fmt.Errorf("failed to run the frontend server: %v", err)
	}

	klog.V(1).Infoln("Starting agent server for tunnel connections.")
	err = p.runAgentServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the agent server: %v", err)
	}
	klog.V(1).Infoln("Starting admin server for debug connections.")
	err = p.runAdminServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the admin server: %v", err)
	}
	klog.V(1).Infoln("Starting health server for healthchecks.")
	err = p.runHealthServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the health server: %v", err)
	}

	stopCh := SetupSignalHandler()
	<-stopCh
	klog.V(1).Infoln("Shutting down server.")

	if frontendStop != nil {
		frontendStop()
	}

	return nil
}

var shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}

func SetupSignalHandler() (stopCh <-chan struct{}) {
	stop := make(chan struct{})
	c := make(chan os.Signal, 2)
	signal.Notify(c, shutdownSignals...)
	go func() {
		<-c
		close(stop)
		<-c
		os.Exit(1) // second signal. Exit directly.
	}()

	return stop
}

func getUDSListener(ctx context.Context, udsName string) (net.Listener, error) {
	udsListenerLock.Lock()
	defer udsListenerLock.Unlock()
	oldUmask := syscall.Umask(0007)
	defer syscall.Umask(oldUmask)
	var lc net.ListenConfig
	lis, err := lc.Listen(ctx, "unix", udsName)
	if err != nil {
		return nil, fmt.Errorf("failed to listen(unix) name %s: %v", udsName, err)
	}
	return lis, nil
}

func (p *Proxy) runFrontendServer(ctx context.Context, o *options.ProxyRunOptions, server *server.ProxyServer) (StopFunc, error) {
	if o.UdsName != "" {
		return p.runUDSFrontendServer(ctx, o, server)
	}
	return p.runMTLSFrontendServer(ctx, o, server)
}

func (p *Proxy) runUDSFrontendServer(ctx context.Context, o *options.ProxyRunOptions, s *server.ProxyServer) (StopFunc, error) {
	if o.DeleteUDSFile {
		if err := os.Remove(o.UdsName); err != nil && !os.IsNotExist(err) {
			klog.ErrorS(err, "failed to delete file", "file", o.UdsName)
		}
	}
	var stop StopFunc
	if o.Mode == "grpc" {
		grpcServer := grpc.NewServer()
		client.RegisterProxyServiceServer(grpcServer, s)
		lis, err := getUDSListener(ctx, o.UdsName)
		if err != nil {<commit_msg>proxy-server: in gRPC mode, set same keepalive for UDS and mTLS.<commit_after>package app

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/http/pprof"
	"os"
	"os/signal"
	"path/filepath"
	"runtime"
	"sync"
	"syscall"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/spf13/cobra"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/keepalive"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/klog/v2"

	"sigs.k8s.io/apiserver-network-proxy/cmd/server/app/options"
	"sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client"
	"sigs.k8s.io/apiserver-network-proxy/pkg/server"
	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
)

var udsListenerLock sync.Mutex

func NewProxyCommand(p *Proxy, o *options.ProxyRunOptions) *cobra.Command {
	cmd := &cobra.Command{
		Use:  "proxy",
		Long: `A gRPC proxy server, receives requests from the API server and forwards to the agent.`,
		RunE: func(cmd *cobra.Command, args []string) error {
			return p.run(o)
		},
	}

	return cmd
}

type Proxy struct {
}

type StopFunc func()

func (p *Proxy) run(o *options.ProxyRunOptions) error {
	o.Print()
	if err := o.Validate(); err != nil {
		return fmt.Errorf("failed to validate server options with %v", err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	var k8sClient *kubernetes.Clientset
	if o.AgentNamespace != "" {
		config, err := clientcmd.BuildConfigFromFlags("", o.KubeconfigPath)
		if err != nil {
			return fmt.Errorf("failed to load kubernetes client config: %v", err)
		}

		if o.KubeconfigQPS != 0 {
			klog.V(1).Infof("Setting k8s client QPS: %v", o.KubeconfigQPS)
			config.QPS = o.KubeconfigQPS
		}
		if o.KubeconfigBurst != 0 {
			klog.V(1).Infof("Setting k8s client Burst: %v", o.KubeconfigBurst)
			config.Burst = o.KubeconfigBurst
		}
		k8sClient, err = kubernetes.NewForConfig(config)
		if err != nil {
			return fmt.Errorf("failed to create kubernetes clientset: %v", err)
		}
	}

	authOpt := &server.AgentTokenAuthenticationOptions{
		Enabled:                o.AgentNamespace != "",
		AgentNamespace:         o.AgentNamespace,
		AgentServiceAccount:    o.AgentServiceAccount,
		KubernetesClient:       k8sClient,
		AuthenticationAudience: o.AuthenticationAudience,
	}
	klog.V(1).Infoln("Starting frontend server for client connections.")
	ps, err := server.GenProxyStrategiesFromStr(o.ProxyStrategies)
	if err != nil {
		return err
	}
	server := server.NewProxyServer(o.ServerID, ps, int(o.ServerCount), authOpt, o.WarnOnChannelLimit)

	frontendStop, err := p.runFrontendServer(ctx, o, server)
	if err != nil {
		return fmt.Errorf("failed to run the frontend server: %v", err)
	}

	klog.V(1).Infoln("Starting agent server for tunnel connections.")
	err = p.runAgentServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the agent server: %v", err)
	}
	klog.V(1).Infoln("Starting admin server for debug connections.")
	err = p.runAdminServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the admin server: %v", err)
	}
	klog.V(1).Infoln("Starting health server for healthchecks.")
	err = p.runHealthServer(o, server)
	if err != nil {
		return fmt.Errorf("failed to run the health server: %v", err)
	}

	stopCh := SetupSignalHandler()
	<-stopCh
	klog.V(1).Infoln("Shutting down server.")

	if frontendStop != nil {
		frontendStop()
	}

	return nil
}

var shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}

func SetupSignalHandler() (stopCh <-chan struct{}) {
	stop := make(chan struct{})
	c := make(chan os.Signal, 2)
	signal.Notify(c, shutdownSignals...)
	go func() {
		<-c
		close(stop)
		<-c
		os.Exit(1) // second signal. Exit directly.
	}()

	return stop
}

func getUDSListener(ctx context.Context, udsName string) (net.Listener, error) {
	udsListenerLock.Lock()
	defer udsListenerLock.Unlock()
	oldUmask := syscall.Umask(0007)
	defer syscall.Umask(oldUmask)
	var lc net.ListenConfig
	lis, err := lc.Listen(ctx, "unix", udsName)
	if err != nil {
		return nil, fmt.Errorf("failed to listen(unix) name %s: %v", udsName, err)
	}
	return lis, nil
}

func (p *Proxy) runFrontendServer(ctx context.Context, o *options.ProxyRunOptions, server *server.ProxyServer) (StopFunc, error) {
	if o.UdsName != "" {
		return p.runUDSFrontendServer(ctx, o, server)
	}
	return p.runMTLSFrontendServer(ctx, o, server)
}

func (p *Proxy) runUDSFrontendServer(ctx context.Context, o *options.ProxyRunOptions, s *server.ProxyServer) (StopFunc, error) {
	if o.DeleteUDSFile {
		if err := os.Remove(o.UdsName); err != nil && !os.IsNotExist(err) {
			klog.ErrorS(err, "failed to delete file", "file", o.UdsName)
		}
	}
	var stop StopFunc
	if o.Mode == "grpc" {
		frontendServerOptions := []grpc.ServerOption{
			grpc.KeepaliveParams(keepalive.ServerParameters{Time: o.FrontendKeepaliveTime}),
		}
		grpcServer := grpc.NewServer(frontendServerOptions...)
		client.RegisterProxyServiceServer(grpcServer, s)
		lis, err := getUDSListener(ctx, o.UdsName)
		if err != nil {
{'fraction_non_alphanumeric': 0.14248678831388972, 'fraction_numerical': 0.0067803370226343605, 'mean_word_length': 4.21038961038961, 'pattern_counts': {'":': 0, '<': 11, '<?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': '5075645', 'n_tokens_mistral': 3950, 'n_tokens_neox': 3739, 'n_words': 840}