<commit_before>package kic

import (
	"fmt"
	"strings"
	"time"

	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	typed_core "k8s.io/client-go/kubernetes/typed/core/v1"

	"k8s.io/minikube/pkg/minikube/tunnel"
)

// Tunnel ...
type Tunnel struct {
	sshPort              string
	sshKey               string
	v1Core               typed_core.CoreV1Interface
	LoadBalancerEmulator tunnel.LoadBalancerEmulator
}

// NewTunnel ...
func NewTunnel(sshPort, sshKey string, v1Core typed_core.CoreV1Interface) *Tunnel {
	return &Tunnel{
		sshPort:              sshPort,
		sshKey:               sshKey,
		v1Core:               v1Core,
		LoadBalancerEmulator: tunnel.NewLoadBalancerEmulator(v1Core),
	}
}

// Start ...
func (t *Tunnel) Start() error {
	sshTunnels := make(map[string]*sshTunnel)

	for {
		services, err := t.v1Core.Services("").List(metav1.ListOptions{})
		if err != nil {
			return err
		}

		for _, sshTunnel := range sshTunnels {
			sshTunnel.sigkill = true
		}

		for _, s := range services.Items {
			if s.Spec.Type == v1.ServiceTypeLoadBalancer {
				name := sshTunnelName(s)
				existingSSHTunnel, ok := sshTunnels[name]

				if ok {
					existingSSHTunnel.sigkill = false
					continue
				}

				newSSHTunnel := createSSHTunnel(name, t.sshPort, t.sshKey, s)
				sshTunnels[newSSHTunnel.name] = newSSHTunnel
				_, err := t.LoadBalancerEmulator.PatchServicesIP("127.0.0.1")
				if err != nil {
					fmt.Println(err)
				}
			}
		}

		for _, sshTunnel := range sshTunnels {
			if sshTunnel.sigkill {
				sshTunnel.stop()
				delete(sshTunnels, sshTunnel.name)
			}
		}

		// TODO: which time to use?
		time.Sleep(1 * time.Second)
	}
}

// sshTunnelName creaets a uniq name for the tunnel, using its name/clusterIP/ports.
// This allows a new process to be created if an existing service was changed,
// the new process will support the IP/Ports change ocurred.
func sshTunnelName(service v1.Service) string {
	n := []string{
		service.Name,
		"-",
		service.Spec.ClusterIP,
	}

	for _, port := range service.Spec.Ports {
		n = append(n, fmt.Sprintf("-%d", port.Port))
	}

	return strings.Join(n, "")
}
<commit_msg>Add cleanup to kic tunnel<commit_after>package kic

import (
	"context"
	"fmt"
	"strings"
	"time"

	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	typed_core "k8s.io/client-go/kubernetes/typed/core/v1"

	"k8s.io/minikube/pkg/minikube/tunnel"
)

// Tunnel ...
type Tunnel struct {
	ctx                  context.Context
	sshPort              string
	sshKey               string
	v1Core               typed_core.CoreV1Interface
	LoadBalancerEmulator tunnel.LoadBalancerEmulator
}

// NewTunnel ...
func NewTunnel(ctx context.Context, sshPort, sshKey string, v1Core typed_core.CoreV1Interface) *Tunnel {
	return &Tunnel{
		ctx:                  ctx,
		sshPort:              sshPort,
		sshKey:               sshKey,
		v1Core:               v1Core,
		LoadBalancerEmulator: tunnel.NewLoadBalancerEmulator(v1Core),
	}
}

// Start ...
func (t *Tunnel) Start() error {
	sshTunnels := make(map[string]*sshTunnel)

	for {
		select {
		case <-t.ctx.Done():
			_, err := t.LoadBalancerEmulator.Cleanup()
			if err != nil {
				fmt.Println(err)
			}
			return nil
		default:
		}

		services, err := t.v1Core.Services("").List(metav1.ListOptions{})
		if err != nil {
			return err
		}

		for _, sshTunnel := range sshTunnels {
			sshTunnel.sigkill = true
		}

		for _, s := range services.Items {
			if s.Spec.Type == v1.ServiceTypeLoadBalancer {
				name := sshTunnelName(s)
				existingSSHTunnel, ok := sshTunnels[name]

				if ok {
					existingSSHTunnel.sigkill = false
					continue
				}

				newSSHTunnel := createSSHTunnel(name, t.sshPort, t.sshKey, s)
				sshTunnels[newSSHTunnel.name] = newSSHTunnel

				// PatchServicesIP maybe can receive the svc here, instead of listing inside again
				_, err := t.LoadBalancerEmulator.PatchServicesIP("127.0.0.1")
				if err != nil {
					fmt.Println(err)
				}
			}
		}

		for _, sshTunnel := range sshTunnels {
			if sshTunnel.sigkill {
				sshTunnel.stop()
				delete(sshTunnels, sshTunnel.name)
			}
		}

		// TODO: which time to use?
		time.Sleep(1 * time.Second)
	}
}

// sshTunnelName creaets a uniq name for the tunnel, using its name/clusterIP/ports.
// This allows a new process to be created if an existing service was changed,
// the new process will support the IP/Ports change ocurred.
func sshTunnelName(service v1.Service) string {
	n := []string{
		service.Name,
		"-",
		service.Spec.ClusterIP,
	}

	for _, port := range service.Spec.Ports {
		n = append(n, fmt.Sprintf("-%d", port.Port))
	}

	return strings.Join(n, "")
}

{'fraction_non_alphanumeric': 0.12139474817046922, 'fraction_numerical': 0.01162290142057684, 'mean_word_length': 2.7719155844155843, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '2343251', 'n_tokens_mistral': 1944, 'n_tokens_neox': 1684, 'n_words': 363}