krt: initial debug interface

istio · istio-testing · Nov 5, 2024 · Oct 21, 2024 · Oct 30, 2024 · Oct 30, 2024
commit 3074fd0fda398f3acbbed794db780d9f59f84d69
@@ -46,6 +46,7 @@ import (
 	"istio.io/istio/pkg/config"
 	"istio.io/istio/pkg/config/schema/resource"
 	"istio.io/istio/pkg/config/xds"
+	"istio.io/istio/pkg/kube/krt"
 	istiolog "istio.io/istio/pkg/log"
 	"istio.io/istio/pkg/security"
 	"istio.io/istio/pkg/slices"
@@ -205,6 +206,7 @@ func (s *DiscoveryServer) AddDebugHandlers(mux, internalMux *http.ServeMux, enab
 	s.addDebugHandler(mux, internalMux, "/debug/resourcesz", "Debug support for watched resources", s.resourcez)
 	s.addDebugHandler(mux, internalMux, "/debug/instancesz", "Debug support for service instances", s.instancesz)
 	s.addDebugHandler(mux, internalMux, "/debug/ambientz", "Debug support for ambient", s.ambientz)
+	s.addDebugHandler(mux, internalMux, "/debug/krtz", "Debug support for krt (internal state)", s.krtz)
 
 	s.addDebugHandler(mux, internalMux, "/debug/authorizationz", "Internal authorization policies", s.authorizationz)
 	s.addDebugHandler(mux, internalMux, "/debug/telemetryz", "Debug Telemetry configuration", s.telemetryz)
@@ -1080,6 +1082,10 @@ func (s *DiscoveryServer) ambientz(w http.ResponseWriter, req *http.Request) {
 	writeJSON(w, res, req)
 }
 
+func (s *DiscoveryServer) krtz(w http.ResponseWriter, req *http.Request) {
+	writeJSON(w, krt.DebugCollections, req)
+}
+
 func (s *DiscoveryServer) networkz(w http.ResponseWriter, req *http.Request) {
 	if s.Env == nil || s.Env.NetworkManager == nil {
 		return

@@ -159,24 +159,54 @@ func (h *manyCollection[I, O]) Synced() Syncer {
 }
 
 // nolint: unused // (not true, its to implement an interface)
-func (h *manyCollection[I, O]) dump() {
+func (h *manyCollection[I, O]) dump() CollectionDump {
 	h.recomputeMu.Lock()
 	defer h.recomputeMu.Unlock()
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	h.log.Errorf(">>> BEGIN DUMP")
+
+	inputs := make(map[string]InputDump, len(h.collectionState.inputs))
+	for k, v := range h.collectionState.mappings {
+		output := make([]string, 0, len(v))
+		for vv := range v {
+			output = append(output, string(vv))
+		}
+		slices.Sort(output)
+		inputs[string(k)] = InputDump{
+			Outputs:      output,
+			Dependencies: nil, // filled later
+		}
+	}
 	for k, deps := range h.objectDependencies {
+		depss := make([]string, 0, len(deps))
 		for _, dep := range deps {
-			h.log.Errorf("Dependencies for: %v: %v (%v)", k, dep.collectionName, dep.filter)
+			depss = append(depss, fmt.Sprintf("%v with filter %v", dep.collectionName, dep.filter))
 		}
-	}
-	for i, os := range h.collectionState.mappings {
-		h.log.Errorf("Input %v -> %v", i, os.UnsortedList())
-	}
-	for os, o := range h.collectionState.outputs {
-		h.log.Errorf("Output %v -> %v", os, o)
-	}
-	h.log.Errorf("<<< END DUMP")
+		slices.Sort(depss)
+		cur := inputs[string(k)]
+		cur.Dependencies = depss
+		inputs[string(k)] = cur
+	}
+	// var xx Key[I]
+	// x := h.objectDependencies[xx][0].
+
+	return CollectionDump{
+		Outputs: eraseMap(h.collectionState.outputs),
+		Inputs:  inputs,
+	}
+	//h.log.Errorf(">>> BEGIN DUMP")
+	//for k, deps := range h.objectDependencies {
+	//	for _, dep := range deps {
+	//		h.log.Errorf("Dependencies for: %v: %v (%v)", k, dep.collectionName, dep.filter)
+	//	}
+	//}
+	//for i, os := range h.collectionState.mappings {
+	//	h.log.Errorf("Input %v -> %v", i, os.UnsortedList())
+	//}
+	//for os, o := range h.collectionState.outputs {
+	//	h.log.Errorf("Output %v -> %v", os, o)
+	//}
+	//h.log.Errorf("<<< END DUMP")
 }
 
 // nolint: unused // (not true, its to implement an interface)
@@ -424,6 +454,7 @@ func newManyCollection[I, O any](cc Collection[I], hf TransformationMulti[I, O],
 		synced:        make(chan struct{}),
 		stop:          opts.stop,
 	}
+	RegisterCollectionForDebugging(h)
 	go func() {
 		// Wait for primary dependency to be ready
 		if !c.Synced().WaitUntilSynced(h.stop) {

@@ -73,7 +73,7 @@ type internalCollection[T any] interface {
 	// Uid is an internal unique ID for this collection. MUST be globally unique
 	uid() collectionUID
 
-	dump()
+	dump() CollectionDump
 
 	// Augment mutates an object for use in various function calls. See WithObjectAugmentation
 	augment(any) any

@@ -0,0 +1,53 @@
+package krt
+
+import (
+	"encoding/json"
+
+	"istio.io/istio/pkg/slices"
+)
+
+var DebugCollections = []DebugCollection{}
+
+type CollectionDump struct {
+	// Map of output key -> output
+	Outputs map[string]any `json:"outputs"`
+	// Map of input key -> info
+	Inputs map[string]InputDump `json:"inputs"`
+}
+type InputDump struct {
+	Outputs      []string `json:"outputs"`
+	Dependencies []string `json:"dependencies"`
+}
+type DebugCollection struct {
+	name string
+	dump func() CollectionDump
+}
+
+func (p DebugCollection) MarshalJSON() ([]byte, error) {
+	return json.Marshal(map[string]any{
+		"name":  p.name,
+		"state": p.dump(),
+	})
+}
+
+func RegisterCollectionForDebugging[T any](c Collection[T]) {
+	cc := c.(internalCollection[T])
+	DebugCollections = append(DebugCollections, DebugCollection{
+		name: cc.name(),
+		dump: cc.dump,
+	})
+}
+
+func eraseSlice[T any](l []T) []any {
+	return slices.Map(l, func(e T) any {
+		return any(e)
+	})
+}
+
+func eraseMap[T any](l map[Key[T]]T) map[string]any {
+	nm := make(map[string]any, len(l))
+	for k, v := range l {
+		nm[string(k)] = v
+	}
+	return nm
+}
@@ -27,6 +27,7 @@ import (
 	"istio.io/istio/pkg/kube/kubetypes"
 	istiolog "istio.io/istio/pkg/log"
 	"istio.io/istio/pkg/ptr"
+	"istio.io/istio/pkg/slices"
 )
 
 type informer[I controllers.ComparableObject] struct {
@@ -60,12 +61,12 @@ func (i *informer[I]) Synced() Syncer {
 }
 
 // nolint: unused // (not true, its to implement an interface)
-func (i *informer[I]) dump() {
-	i.log.Errorf(">>> BEGIN DUMP")
-	for _, obj := range i.inf.List(metav1.NamespaceAll, klabels.Everything()) {
-		i.log.Errorf("%s/%s", obj.GetNamespace(), obj.GetName())
+func (i *informer[I]) dump() CollectionDump {
+	return CollectionDump{
+		Outputs: eraseMap(slices.GroupUnique(i.inf.List(metav1.NamespaceAll, klabels.Everything()), func(t I) Key[I] {
+			return GetKey(t)
+		})),
 	}
-	i.log.Errorf("<<< END DUMP")
 }
 
 func (i *informer[I]) name() string {
@@ -184,6 +185,7 @@ func WrapClient[I controllers.ComparableObject](c kclient.Informer[I], opts ...C
 		close(h.synced)
 		h.log.Infof("%v synced", h.name())
 	}()
+	RegisterCollectionForDebugging(h)
 	return h
 }
 

@@ -79,12 +79,14 @@ func (j *join[T]) name() string { return j.collectionName }
 func (j *join[T]) uid() collectionUID { return j.id }
 
 // nolint: unused // (not true, its to implement an interface)
-func (j *join[I]) dump() {
+func (j *join[I]) dump() CollectionDump {
 	log.Errorf("> BEGIN DUMP (join %v)", j.collectionName)
-	for _, c := range j.collections {
-		c.dump()
-	}
+	//for _, c := range j.collections {
+	//	c.dump()
+	//}
+	// TODO
 	log.Errorf("< END DUMP (join %v)", j.collectionName)
+	return CollectionDump{}
 }
 
 // nolint: unused // (not true)

@@ -106,8 +106,9 @@ func (d *static[T]) Set(now *T) {
 }
 
 // nolint: unused // (not true, its to implement an interface)
-func (d *static[T]) dump() {
+func (d *static[T]) dump() CollectionDump {
 	log.Errorf(">>> static[%v]: %+v<<<", ptr.TypeName[T](), d.val.Load())
+	return CollectionDump{}
 }
 
 // nolint: unused // (not true, its to implement an interface)

@@ -122,7 +122,8 @@ func (s *staticList[T]) uid() collectionUID {
 }
 
 // nolint: unused // (not true, its to implement an interface)
-func (s *staticList[T]) dump() {
+func (s *staticList[T]) dump() CollectionDump {
+	return CollectionDump{}
 }
 
 // nolint: unused // (not true, its to implement an interface)