[DO NOT MERGE] Demo for Cilium dev day

demo/handlers.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"text/tabwriter"
+	"time"
+
+	"github.com/cilium/statedb"
+	v1 "k8s.io/api/core/v1"
+)
+
+func registerStateDBHTTPHandler(mux *http.ServeMux, db *statedb.DB) {
+	mux.Handle("/statedb", db)
+}
+
+func registerPodHTTPHandler(mux *http.ServeMux, db *statedb.DB, pods statedb.Table[*Pod]) {
+	mux.HandleFunc("/pods/running", func(w http.ResponseWriter, req *http.Request) {
+		txn := db.ReadTxn()
+		iter, _ := pods.Get(txn, PodPhaseIndex.Query(v1.PodRunning))
+		t := tabwriter.NewWriter(w, 10, 4, 2, ' ', 0)
+		fmt.Fprintf(t, "NAME\tSTARTED\tSTATUS\n")
+		for pod, _, ok := iter.Next(); ok; pod, _, ok = iter.Next() {
+			fmt.Fprintf(t, "%s/%s\t%s ago\t\t%s\n", pod.Namespace, pod.Name, time.Since(pod.StartTime), pod.ReconciliationStatus())
+		}
+		t.Flush()
+	})
+}

demo/main.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"log/slog"
+	"net/http"
+	"path"
+	"time"
+
+	"github.com/cilium/hive"
+	"github.com/cilium/hive/cell"
+	"github.com/cilium/hive/job"
+	"github.com/cilium/statedb"
+	"github.com/cilium/statedb/reconciler"
+	"github.com/cilium/statedb/reflector"
+	"github.com/spf13/cobra"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/homedir"
+)
+
+var Hive = hive.New(
+	job.Cell,
+	statedb.Cell,
+	cell.SimpleHealthCell,
+
+	// Kubernetes client
+	cell.Provide(
+		newClientset,
+	),
+
+	// HTTP server
+	cell.Provide(
+		http.NewServeMux,
+	),
+	cell.Invoke(
+		registerHTTPServer,
+		registerStateDBHTTPHandler,
+	),
+
+	// Pod tables and the reconciler
+	cell.Provide(
+		NewPodTable,
+		statedb.RWTable[*Pod].ToTable,
+		podReflectorConfig,
+		podReconcilerConfig,
+	),
+
+	reflector.KubernetesCell[*Pod](),
+
+	cell.Invoke(
+		statedb.RegisterTable[*Pod],
+
+		reconciler.Register[*Pod],
+
+		registerPodHTTPHandler,
+	),
+)
+
+var cmd = &cobra.Command{
+	Use: "example",
+	RunE: func(_ *cobra.Command, args []string) error {
+		return Hive.Run()
+	},
+}
+
+func main() {
+	// Register all configuration flags in the hive to the command
+	Hive.RegisterFlags(cmd.Flags())
+
+	// Add the "hive" sub-command for inspecting the hive
+	cmd.AddCommand(Hive.Command())
+
+	// And finally execute the command to parse the command-line flags and
+	// run the hive
+	cmd.Execute()
+}
+
+func podReflectorConfig(client *kubernetes.Clientset, pods statedb.RWTable[*Pod]) reflector.KubernetesConfig[*Pod] {
+	lw := ListerWatcherFromTyped(client.CoreV1().Pods(""))
+	return reflector.KubernetesConfig[*Pod]{
+		BufferSize:     100,
+		BufferWaitTime: 100 * time.Millisecond,
+		ListerWatcher:  lw,
+		Table:          pods,
+		Transform: func(obj any) (*Pod, bool) {
+			pod, ok := obj.(*v1.Pod)
+			if ok {
+				return fromV1Pod(pod), true
+			}
+			return nil, false
+		},
+	}
+}
+
+func newClientset() (*kubernetes.Clientset, error) {
+	cfg, err := clientcmd.BuildConfigFromFlags("", path.Join(homedir.HomeDir(), ".kube", "config"))
+	if err != nil {
+		panic(err.Error())
+	}
+
+	return kubernetes.NewForConfig(cfg)
+}
+
+func registerHTTPServer(log *slog.Logger, mux *http.ServeMux, lc cell.Lifecycle) {
+	s := &http.Server{Addr: ":8080", Handler: mux}
+	lc.Append(cell.Hook{
+		OnStart: func(cell.HookContext) error {
+			log.Info("Serving HTTP", "addr", s.Addr)
+			go s.ListenAndServe()
+			return nil
+		},
+		OnStop: func(ctx cell.HookContext) error {
+			return s.Shutdown(ctx)
+		},
+	})
+
+}

demo/reconciler.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/cilium/statedb"
+	"github.com/cilium/statedb/reconciler"
+)
+
+type podOps struct {
+	log *slog.Logger
+}
+
+// Delete implements reconciler.Operations.
+func (o *podOps) Delete(_ context.Context, _ statedb.ReadTxn, pod *Pod) error {
+	o.log.Info("Pod deleted", "name", pod.Namespace+"/"+pod.Name)
+	return nil
+}
+
+// Prune implements reconciler.Operations.
+func (o *podOps) Prune(context.Context, statedb.ReadTxn, statedb.Iterator[*Pod]) error {
+	return nil
+}
+
+// Update implements reconciler.Operations.
+func (o *podOps) Update(ctx context.Context, txn statedb.ReadTxn, pod *Pod, changed *bool) error {
+	o.log.Info("Pod updated", "name", pod.Namespace+"/"+pod.Name, "phase", pod.Phase)
+	return nil
+}
+
+var _ reconciler.Operations[*Pod] = &podOps{}
+
+func podReconcilerConfig(log *slog.Logger) reconciler.Config[*Pod] {
+	return reconciler.Config[*Pod]{
+		FullReconcilationInterval: time.Minute,
+		RetryBackoffMinDuration:   100 * time.Millisecond,
+		RetryBackoffMaxDuration:   time.Minute,
+		IncrementalRoundSize:      1000,
+		GetObjectStatus:           (*Pod).ReconciliationStatus,
+		WithObjectStatus:          (*Pod).WithReconciliationStatus,
+		Operations:                &podOps{log},
+	}
+}

demo/tables.go

@@ -0,0 +1,71 @@
+package main
+
+import (
+	"time"
+
+	v1 "k8s.io/api/core/v1"
+
+	"github.com/cilium/statedb"
+	"github.com/cilium/statedb/index"
+	"github.com/cilium/statedb/reconciler"
+)
+
+type Pod struct {
+	Name, Namespace string
+	Phase           v1.PodPhase
+	StartTime       time.Time
+
+	reconciliationStatus reconciler.Status
+}
+
+func fromV1Pod(p *v1.Pod) *Pod {
+	return &Pod{
+		Name:                 p.Name,
+		Namespace:            p.Namespace,
+		Phase:                p.Status.Phase,
+		StartTime:            p.Status.StartTime.Time,
+		reconciliationStatus: reconciler.StatusPending(),
+	}
+}
+
+func (p *Pod) ReconciliationStatus() reconciler.Status {
+	return p.reconciliationStatus
+}
+
+func (p *Pod) WithReconciliationStatus(s reconciler.Status) *Pod {
+	p2 := *p
+	p2.reconciliationStatus = s
+	return &p2
+}
+
+const PodTableName = "pods"
+
+var (
+	PodNameIndex = statedb.Index[*Pod, string]{
+		Name: "name",
+		FromObject: func(pod *Pod) index.KeySet {
+			return index.NewKeySet(index.String(pod.Namespace + "/" + pod.Name))
+		},
+		FromKey: index.String,
+		Unique:  true,
+	}
+	PodPhaseIndex = statedb.Index[*Pod, v1.PodPhase]{
+		Name: "phase",
+		FromObject: func(pod *Pod) index.KeySet {
+			return index.NewKeySet(index.String(string(pod.Phase)))
+		},
+		FromKey: func(key v1.PodPhase) index.Key {
+			return index.String(string(key))
+		},
+		Unique: false,
+	}
+)
+
+func NewPodTable(db *statedb.DB) (statedb.RWTable[*Pod], error) {
+	return statedb.NewTable[*Pod](
+		PodTableName,
+
+		PodNameIndex,
+		PodPhaseIndex,
+	)
+}

demo/utils.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/tools/cache"
+)
+
+// typedListWatcher is a generic interface that all the typed k8s clients match.
+type typedListWatcher[T runtime.Object] interface {
+	List(ctx context.Context, opts metav1.ListOptions) (T, error)
+	Watch(ctx context.Context, opts metav1.ListOptions) (watch.Interface, error)
+}
+
+// genListWatcher takes a typed list watcher and implements cache.ListWatch
+// using it.
+type genListWatcher[T runtime.Object] struct {
+	lw typedListWatcher[T]
+}
+
+func (g *genListWatcher[T]) List(opts metav1.ListOptions) (runtime.Object, error) {
+	return g.lw.List(context.Background(), opts)
+}
+
+func (g *genListWatcher[T]) Watch(opts metav1.ListOptions) (watch.Interface, error) {
+	return g.lw.Watch(context.Background(), opts)
+}
+
+// ListerWatcherFromTyped adapts a typed k8s client to cache.ListerWatcher so it can be used
+// with an informer. With this construction we can use fake clients for testing,
+// which would not be possible if we used NewListWatchFromClient and RESTClient().
+func ListerWatcherFromTyped[T runtime.Object](lw typedListWatcher[T]) cache.ListerWatcher {
+	return &genListWatcher[T]{lw: lw}
+}

go.mod

@@ -19,23 +19,30 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-openapi/jsonpointer v0.20.0 // indirect
+	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
+	github.com/imdario/mergo v0.3.6 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/magiconair/properties v1.8.7 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
@@ -59,6 +66,7 @@ require (
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/klog/v2 v2.120.1 // indirect
+	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
 	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect