Etcd wal decoder (kubernetes lease only) example, used by https://github.com/etcd-io/etcd/tree/v3.5.9/tools/etcd-dump-logs

etcd-wal-decoder-kubernetes.go

package main

import (
	"bufio"
	"bytes"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"os"

	"go.etcd.io/etcd/api/v3/etcdserverpb"
	v1 "k8s.io/api/coordination/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes/scheme"
)

const (
	decodeStatusOK    = "OK"
	decodeStatusError = "ERROR"

	targetPrefix = "/registry/leases/"
)

var (
	codec   = scheme.Codecs
	decoder = codec.UniversalDecoder()
	newObj  = func() runtime.Object { return &v1.Lease{} }
)

var (
	in  io.Reader = os.Stdin
	out io.Writer = os.Stdout
)

func main() {
	scanner := bufio.NewScanner(in)

	for scanner.Scan() {
		line := scanner.Text()
		data, err := decode(line)
		if err != nil {
			fmt.Fprintf(out, "%s|%s\n", decodeStatusError, err)
		} else {
			fmt.Fprintf(out, "%s|%s\n", decodeStatusOK, data)
		}
	}
}

func decode(line string) ([]byte, error) {
	input, err := hex.DecodeString(line)
	if err != nil {
		return nil, fmt.Errorf("failed to decode hex string: %w", err)
	}

	var rr etcdserverpb.InternalRaftRequest
	if err := rr.Unmarshal(input); err != nil {
		return nil, fmt.Errorf("failed to unmarshal raft request: %w", err)
	}
	if rr.Txn == nil {
		return nil, fmt.Errorf("no transaction found")
	}
	if len(rr.Txn.Success) != 1 {
		return nil, fmt.Errorf("expected 1 success response, got %d", len(rr.Txn.Success))
	}

	put := rr.Txn.Success[0].GetRequestPut()
	if put == nil {
		return nil, fmt.Errorf("no put request found")
	}
	if !bytes.HasPrefix(put.Key, []byte(targetPrefix)) {
		return nil, fmt.Errorf("key does not have target prefix")
	}
	obj, _, err := decoder.Decode(put.Value, nil, newObj())
	if err != nil {
		return nil, fmt.Errorf("failed to decode resource: %w", err)
	}
	out, err := json.Marshal(obj)
	if err != nil {
		return nil, fmt.Errorf("json marshal error: %v", err)
	}

	return out, nil
}