<commit_before>package watcher

import (
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/datawire/teleproxy/internal/pkg/k8s"

	"k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/watch"

	"k8s.io/client-go/discovery"
	"k8s.io/client-go/dynamic"
	_ "k8s.io/client-go/plugin/pkg/client/auth"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
)

type empty struct{}

type listWatchAdapter struct {
	resource dynamic.NamespaceableResourceInterface
}

func (lw listWatchAdapter) List(options v1.ListOptions) (runtime.Object, error) {
	return lw.resource.List(options)
}

func (lw listWatchAdapter) Watch(options v1.ListOptions) (watch.Interface, error) {
	return lw.resource.Watch(options)
}

type Watcher struct {
	config       *rest.Config
	resources    []*v1.APIResourceList
	stores       map[string]cache.Store
	stop         chan empty
	stopChans    []chan struct{}
	stoppedChans []chan empty
}

// NewWatcher returns a Kubernetes Watcher for the specified cluster
func NewWatcher(kubeinfo *k8s.KubeInfo) *Watcher {
	config, err := kubeinfo.GetRestConfig()
	if err != nil {
		log.Fatalln("Failed to get REST config:", err)
	}

	disco, err := discovery.NewDiscoveryClientForConfig(config)
	if err != nil {
		log.Fatal(err)
	}

	resources, err := disco.ServerResources()
	if err != nil {
		log.Fatal(err)
	}

	w := &Watcher{
		config:    config,
		resources: resources,
		stores:    make(map[string]cache.Store),
		stop:      make(chan empty),
	}

	go func() {
		<-w.stop
		for _, c := range w.stopChans {
			close(c)
		}
		for {
			<-w.stop
		}
	}()

	return w
}

func (w *Watcher) resolve(resource string) (string, string, v1.APIResource) {
	resource = strings.ToLower(resource)
	if resource == "" {
		return "", "", v1.APIResource{}
	}
	for _, rl := range w.resources {
		for _, r := range rl.APIResources {
			candidates := []string{
				r.Name,
				r.Kind,
				r.SingularName,
			}
			candidates = append(candidates, r.ShortNames...)

			for _, c := range candidates {
				if resource == strings.ToLower(c) {
					var group string
					var version string
					parts := strings.Split(rl.GroupVersion, "/")
					switch len(parts) {
					case 1:
						group = ""
						version = parts[0]
					case 2:
						group = parts[0]
						version = parts[1]
					default:
						panic("unrecognized GroupVersion")
					}
					return group, version, r
				}
			}
		}
	}
	return "", "", v1.APIResource{}
}

func (w *Watcher) Canonical(name string) string {
	parts := strings.Split(name, "/")

	var kind string
	switch len(parts) {
	case 1:
		kind = parts[0]
		name = ""
	case 2:
		kind = parts[0]
		name = parts[1]
	default:
		return ""
	}

	var namespace string
	parts = strings.Split(name, ".")
	switch len(parts) {
	case 0:
	case 1:
		namespace = "default"
	case 2:
		name = parts[0]
		namespace = parts[1]
	default:
		return ""
	}

	_, _, res := w.resolve(kind)
	kind = strings.ToLower(res.Kind)

	if name == "" {
		return kind
	}

	if res.Kind == "Namespace" {
		switch namespace {
		case "":
			fallthrough
		case "default":
			return fmt.Sprintf("%s/%s", kind, name)
		default:
			return ""
		}
	} else {
		return fmt.Sprintf("%s/%s.%s", kind, name, namespace)
	}
}

func (w *Watcher) Watch(resources string, listener func(*Watcher)) error {
	group, version, res := w.resolve(resources)
	if version == "" {
		return fmt.Errorf("unknown resource: '%s'", resources)<commit_msg>fix watching of all non-namespaced resources<commit_after>package watcher

import (
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/datawire/teleproxy/internal/pkg/k8s"

	"k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/watch"

	"k8s.io/client-go/discovery"
	"k8s.io/client-go/dynamic"
	_ "k8s.io/client-go/plugin/pkg/client/auth"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/cache"
)

type empty struct{}

type listWatchAdapter struct {
	resource dynamic.NamespaceableResourceInterface
}

func (lw listWatchAdapter) List(options v1.ListOptions) (runtime.Object, error) {
	return lw.resource.List(options)
}

func (lw listWatchAdapter) Watch(options v1.ListOptions) (watch.Interface, error) {
	return lw.resource.Watch(options)
}

type Watcher struct {
	config       *rest.Config
	resources    []*v1.APIResourceList
	stores       map[string]cache.Store
	stop         chan empty
	stopChans    []chan struct{}
	stoppedChans []chan empty
}

// NewWatcher returns a Kubernetes Watcher for the specified cluster
func NewWatcher(kubeinfo *k8s.KubeInfo) *Watcher {
	config, err := kubeinfo.GetRestConfig()
	if err != nil {
		log.Fatalln("Failed to get REST config:", err)
	}

	disco, err := discovery.NewDiscoveryClientForConfig(config)
	if err != nil {
		log.Fatal(err)
	}

	resources, err := disco.ServerResources()
	if err != nil {
		log.Fatal(err)
	}

	w := &Watcher{
		config:    config,
		resources: resources,
		stores:    make(map[string]cache.Store),
		stop:      make(chan empty),
	}

	go func() {
		<-w.stop
		for _, c := range w.stopChans {
			close(c)
		}
		for {
			<-w.stop
		}
	}()

	return w
}

func (w *Watcher) resolve(resource string) (string, string, v1.APIResource) {
	resource = strings.ToLower(resource)
	if resource == "" {
		return "", "", v1.APIResource{}
	}
	for _, rl := range w.resources {
		for _, r := range rl.APIResources {
			candidates := []string{
				r.Name,
				r.Kind,
				r.SingularName,
			}
			candidates = append(candidates, r.ShortNames...)

			for _, c := range candidates {
				if resource == strings.ToLower(c) {
					var group string
					var version string
					parts := strings.Split(rl.GroupVersion, "/")
					switch len(parts) {
					case 1:
						group = ""
						version = parts[0]
					case 2:
						group = parts[0]
						version = parts[1]
					default:
						panic("unrecognized GroupVersion")
					}
					return group, version, r
				}
			}
		}
	}
	return "", "", v1.APIResource{}
}

func (w *Watcher) Canonical(name string) string {
	parts := strings.Split(name, "/")

	var kind string
	switch len(parts) {
	case 1:
		kind = parts[0]
		name = ""
	case 2:
		kind = parts[0]
		name = parts[1]
	default:
		return ""
	}

	var namespace string
	parts = strings.Split(name, ".")
	switch len(parts) {
	case 0:
	case 1:
		namespace = "default"
	case 2:
		name = parts[0]
		namespace = parts[1]
	default:
		return ""
	}

	_, _, res := w.resolve(kind)
	kind = strings.ToLower(res.Kind)

	if name == "" {
		return kind
	}

	if res.Namespaced {
		return fmt.Sprintf("%s/%s.%s", kind, name, namespace)
	} else {
		switch namespace {
		case "":
			fallthrough
		case "default":
			return fmt.Sprintf("%s/%s", kind, name)
		default:
			return ""
		}
	}
}

func (w *Watcher) Watch(resources string, listener func(*Watcher)) error {
	group, version, res := w.resolve(resources)
	if version == "" {
		return fmt.Errorf("unknown resource: '%s'", resources)
{'fraction_non_alphanumeric': 0.14490084985835694, 'fraction_numerical': 0.009915014164305949, 'mean_word_length': 3.2205618649133294, 'pattern_counts': {'":': 4, '<': 7, '<?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': '4607711', 'n_tokens_mistral': 2914, 'n_tokens_neox': 2588, 'n_words': 575}