Remove some examples

Signed-off-by: Dhi Aurrahman <dio@rockybars.com>

Author: dio

e2e/filters/grpc_callout.go

@@ -0,0 +1,55 @@
+package filters
+
+import (
+	"fmt"
+
+	"github.com/dio/transit/e2e/internal/grpctestproto"
+	"github.com/dio/transit/up"
+)
+
+func init() {
+	up.Register("e2e-grpc-callout", grpcCalloutBodyHeaders, up.WithMutableBody(grpcCalloutBodyHandler))
+}
+
+func grpcCalloutBodyHeaders(_ *up.Writer, r *up.Request) {
+	*r.Context = r.Path
+}
+
+func grpcCalloutBodyHandler(w *up.Writer, chunk *up.BodyChunk) {
+	if !chunk.EndStream {
+		return
+	}
+	req := grpctestproto.EchoRequest{
+		Text:     string(chunk.Data),
+		Sequence: uint32(len(chunk.Data)),
+	}
+	initResult, err := w.GRPCCallout(up.GRPCCalloutRequest{
+		Cluster:       "grpc-callout-upstream",
+		Method:        "/e2e.Echo/Echo",
+		Message:       req.MarshalProto(nil),
+		TimeoutMillis: 1000,
+	}, func(resp up.GRPCCalloutResponse) {
+		if resp.Result != up.HTTPCalloutSuccess {
+			w.SendLocalResponse(503, []byte(fmt.Sprintf("grpc callout failed result=%d", resp.Result)), [2]string{"content-type", "text/plain"})
+			return
+		}
+		if resp.GRPCStatus != 0 {
+			w.SendLocalResponse(502, []byte(fmt.Sprintf("grpc error %d: %s", resp.GRPCStatus, resp.GRPCMessage)),
+				[2]string{"content-type", "text/plain"})
+			return
+		}
+		var echoResp grpctestproto.EchoResponse
+		if err := echoResp.UnmarshalProto(resp.Body); err != nil {
+			w.SendLocalResponse(502, []byte(fmt.Sprintf("grpc decode error: %s", err.Error())), [2]string{"content-type", "text/plain"})
+			return
+		}
+		w.SendLocalResponse(200, []byte(echoResp.Text),
+			[2]string{"content-type", "text/plain"},
+			[2]string{"x-grpc-status", "0"},
+			[2]string{"x-grpc-sequence", fmt.Sprintf("%d", echoResp.Sequence)},
+		)
+	})
+	if err != nil {
+		w.SendLocalResponse(503, []byte(fmt.Sprintf("grpc init=%d err=%s", initResult, err.Error())), [2]string{"content-type", "text/plain"})
+	}
+}

e2e/go.mod

@@ -3,6 +3,7 @@ module github.com/dio/transit/e2e
 go 1.26.2
 
 require (
+	github.com/VictoriaMetrics/easyproto v1.2.0
 	github.com/dio/transit v0.0.0
 	github.com/envoyproxy/envoy/source/extensions/dynamic_modules v0.0.0-20260521055639-0d6e3c60aa55
 	github.com/envoyproxy/go-control-plane/envoy v1.37.0

e2e/go.sum

@@ -1,3 +1,5 @@
+github.com/VictoriaMetrics/easyproto v1.2.0 h1:FJT9uNXA2isppFuJErbLqD306KoFlehl7Wn2dg/6oIE=
+github.com/VictoriaMetrics/easyproto v1.2.0/go.mod h1:QlGlzaJnDfFd8Lk6Ci/fuLxfTo3/GThPs2KH23mv710=
 github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
 github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=

e2e/grpc_callout_test.go

@@ -0,0 +1,62 @@
+package e2e
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/suite"
+)
+
+type GRPCCalloutSuite struct {
+	suite.Suite
+}
+
+func TestGRPCCallout(t *testing.T) {
+	suite.Run(t, new(GRPCCalloutSuite))
+}
+
+func (s *GRPCCalloutSuite) TestPost_echoesBodyViaGRPC() {
+	// The filter reads the request body, encodes it as EchoRequest, issues a
+	// GRPCCallout to the e2e gRPC upstream, decodes the EchoResponse, and
+	// returns the echoed payload as the local response.
+	req, err := http.NewRequest(http.MethodPost, grpcCalloutAddr+"/echo", strings.NewReader("hello grpc"))
+	s.Require().NoError(err)
+	req.Header.Set("content-type", "text/plain")
+
+	resp := mustDo(s.T(), req)
+	body := readBody(s.T(), resp)
+
+	s.Require().Equal(http.StatusOK, resp.StatusCode, "body: %s", body)
+	s.Require().Equal("0", resp.Header.Get("x-grpc-status"))
+	s.Require().Equal("11", resp.Header.Get("x-grpc-sequence"))
+	s.Require().Equal("hello grpc", body)
+}
+
+func (s *GRPCCalloutSuite) TestPost_emptyBody() {
+	req, err := http.NewRequest(http.MethodPost, grpcCalloutAddr+"/echo", strings.NewReader(""))
+	s.Require().NoError(err)
+	req.Header.Set("content-type", "text/plain")
+
+	resp := mustDo(s.T(), req)
+	body := readBody(s.T(), resp)
+
+	s.Require().Equal(http.StatusOK, resp.StatusCode)
+	s.Require().Equal("0", resp.Header.Get("x-grpc-status"))
+	s.Require().Equal("1", resp.Header.Get("x-grpc-sequence"))
+	s.Require().Equal("", body)
+}
+
+func (s *GRPCCalloutSuite) TestPost_largeBody() {
+	payload := strings.Repeat("x", 1024)
+	req, err := http.NewRequest(http.MethodPost, grpcCalloutAddr+"/echo", strings.NewReader(payload))
+	s.Require().NoError(err)
+	req.Header.Set("content-type", "text/plain")
+
+	resp := mustDo(s.T(), req)
+	body := readBody(s.T(), resp)
+
+	s.Require().Equal(http.StatusOK, resp.StatusCode)
+	s.Require().Equal("1025", resp.Header.Get("x-grpc-sequence"))
+	s.Require().Equal(payload, body)
+}

e2e/internal/grpctestproto/proto.go

@@ -0,0 +1,99 @@
+package grpctestproto
+
+import (
+	"fmt"
+
+	"github.com/VictoriaMetrics/easyproto"
+)
+
+// EchoRequest is the proto payload sent by the e2e GRPCCallout filter.
+//
+//	message EchoRequest {
+//	  string text = 1;
+//	  uint32 sequence = 2;
+//	}
+type EchoRequest struct {
+	Text     string
+	Sequence uint32
+}
+
+func (r EchoRequest) MarshalProto(dst []byte) []byte {
+	var m easyproto.Marshaler
+	mm := m.MessageMarshaler()
+	mm.AppendString(1, r.Text)
+	mm.AppendUint32(2, r.Sequence)
+	return m.Marshal(dst)
+}
+
+func (r *EchoRequest) UnmarshalProto(src []byte) error {
+	*r = EchoRequest{}
+	var fc easyproto.FieldContext
+	for len(src) > 0 {
+		var err error
+		src, err = fc.NextField(src)
+		if err != nil {
+			return fmt.Errorf("cannot read EchoRequest field: %w", err)
+		}
+		switch fc.FieldNum {
+		case 1:
+			text, ok := fc.String()
+			if !ok {
+				return fmt.Errorf("cannot read EchoRequest.text")
+			}
+			r.Text = text
+		case 2:
+			sequence, ok := fc.Uint32()
+			if !ok {
+				return fmt.Errorf("cannot read EchoRequest.sequence")
+			}
+			r.Sequence = sequence
+		}
+	}
+	return nil
+}
+
+// EchoResponse is the proto payload returned by the e2e gRPC upstream.
+//
+//	message EchoResponse {
+//	  string text = 1;
+//	  uint32 sequence = 2;
+//	}
+type EchoResponse struct {
+	Text     string
+	Sequence uint32
+}
+
+func (r EchoResponse) MarshalProto(dst []byte) []byte {
+	var m easyproto.Marshaler
+	mm := m.MessageMarshaler()
+	mm.AppendString(1, r.Text)
+	mm.AppendUint32(2, r.Sequence)
+	return m.Marshal(dst)
+}
+
+func (r *EchoResponse) UnmarshalProto(src []byte) error {
+	*r = EchoResponse{}
+	var fc easyproto.FieldContext
+	for len(src) > 0 {
+		var err error
+		src, err = fc.NextField(src)
+		if err != nil {
+			return fmt.Errorf("cannot read EchoResponse field: %w", err)
+		}
+		switch fc.FieldNum {
+		case 1:
+			text, ok := fc.String()
+			if !ok {
+				return fmt.Errorf("cannot read EchoResponse.text")
+			}
+			r.Text = text
+		case 2:
+			sequence, ok := fc.Uint32()
+			if !ok {
+				return fmt.Errorf("cannot read EchoResponse.sequence")
+			}
+			r.Sequence = sequence
+		}
+	}
+	return nil
+}

e2e/main_test.go

@@ -27,6 +27,7 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	_ "embed"
+	"encoding/binary"
 	"encoding/pem"
 	"fmt"
 	"io"
@@ -43,6 +44,7 @@ import (
 	"text/template"
 	"time"
 
+	"github.com/dio/transit/e2e/internal/grpctestproto"
 	"github.com/dio/transit/e2e/sinks/accessloggersink"
 	"github.com/dio/transit/e2e/sinks/alssink"
 	"github.com/dio/transit/e2e/sinks/otelsink"
@@ -75,6 +77,7 @@ var (
 	asyncCalloutBodyAddr          string
 	asyncCalloutLocalResponseAddr string
 	mutableBodyUpstreamAddr       string
+	grpcCalloutAddr               string
 	lbPolicySelectionAddr         string
 	accessLoggerLocalReplyAddr    string
 	accessLoggerFlagsAddr         string
@@ -143,6 +146,8 @@ func TestMain(m *testing.M) {
 	mutableBodyUpstreamPort := freePort()
 	mutableBodyRecorder = startRecorderUpstream()
 	asyncCalloutLocalResponsePort := freePort()
+	grpcCalloutPort := freePort()
+	grpcCalloutUpstreamPort := startGRPCCalloutUpstream()
 	lbPolicySelectionPort := freePort()
 	lbPolicyHost0Port := startIdentifiedUpstream("lb-host-0")
 	lbPolicyHost1Port := startIdentifiedUpstream("lb-host-1")
@@ -184,6 +189,7 @@ func TestMain(m *testing.M) {
 	asyncCalloutBodyAddr = fmt.Sprintf("http://localhost:%d", asyncCalloutBodyPort)
 	mutableBodyUpstreamAddr = fmt.Sprintf("http://localhost:%d", mutableBodyUpstreamPort)
 	asyncCalloutLocalResponseAddr = fmt.Sprintf("http://localhost:%d", asyncCalloutLocalResponsePort)
+	grpcCalloutAddr = fmt.Sprintf("http://localhost:%d", grpcCalloutPort)
 	lbPolicySelectionAddr = fmt.Sprintf("http://localhost:%d", lbPolicySelectionPort)
 	accessLoggerLocalReplyAddr = fmt.Sprintf("http://localhost:%d", accessLoggerLocalReplyPort)
 	accessLoggerFlagsAddr = fmt.Sprintf("http://localhost:%d", accessLoggerFlagsPort)
@@ -272,6 +278,8 @@ func TestMain(m *testing.M) {
 		MutableBodyUpstreamPort:            mutableBodyUpstreamPort,
 		MutableBodyRecorderPort:            mutableBodyRecorder.port,
 		AsyncCalloutLocalResponsePort:      asyncCalloutLocalResponsePort,
+		GRPCCalloutPort:                    grpcCalloutPort,
+		GRPCCalloutUpstreamPort:            grpcCalloutUpstreamPort,
 		LbPolicySelectionPort:              lbPolicySelectionPort,
 		LbPolicyHost0Port:                  lbPolicyHost0Port,
 		LbPolicyHost1Port:                  lbPolicyHost1Port,
@@ -416,6 +424,62 @@ func startAsyncCalloutUpstream() int {
 	return l.Addr().(*net.TCPAddr).Port
 }
 
+// startGRPCCalloutUpstream starts a minimal h2c gRPC server for the GRPCCallout
+// e2e tests. It handles /e2e.Echo/Echo by decoding a framed EchoRequest proto
+// and returning a framed EchoResponse proto with grpc-status: 0 in the trailers.
+func startGRPCCalloutUpstream() int {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		panic("startGRPCCalloutUpstream: " + err.Error())
+	}
+	mux := http.NewServeMux()
+	mux.HandleFunc("/e2e.Echo/Echo", func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		if len(body) < 5 || body[0] != 0 {
+			http.Error(w, "invalid grpc frame", http.StatusBadRequest)
+			return
+		}
+		msgLen := binary.BigEndian.Uint32(body[1:5])
+		end := 5 + int(msgLen)
+		if end < 5 || end > len(body) {
+			http.Error(w, "truncated grpc frame", http.StatusBadRequest)
+			return
+		}
+		var echoReq grpctestproto.EchoRequest
+		if err := echoReq.UnmarshalProto(body[5:end]); err != nil {
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+		echoResp := grpctestproto.EchoResponse{
+			Text:     echoReq.Text,
+			Sequence: echoReq.Sequence + 1,
+		}
+		msg := echoResp.MarshalProto(nil)
+		frame := make([]byte, 5+len(msg))
+		binary.BigEndian.PutUint32(frame[1:5], uint32(len(msg)))
+		copy(frame[5:], msg)
+
+		w.Header().Set("Content-Type", "application/grpc+proto")
+		w.Header().Set("Trailer", "Grpc-Status")
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write(frame)
+		w.Header().Set("Grpc-Status", "0")
+	})
+	protocols := new(http.Protocols)
+	protocols.SetHTTP1(true)
+	protocols.SetUnencryptedHTTP2(true)
+	srv := &http.Server{
+		Handler:   mux,
+		Protocols: protocols,
+	}
+	go srv.Serve(l) //nolint:errcheck
+	return l.Addr().(*net.TCPAddr).Port
+}
+
 // startForwardEchoUpstream starts an upstream that echoes received request
 // headers as "x-received-<lowercase-name>" response headers and returns the
 // request body. Used to verify that filter mutations reached upstream.
@@ -800,6 +864,8 @@ type envoyPorts struct {
 	MutableBodyUpstreamPort            int
 	MutableBodyRecorderPort            int
 	AsyncCalloutLocalResponsePort      int
+	GRPCCalloutPort                    int
+	GRPCCalloutUpstreamPort            int
 	LbPolicySelectionPort              int
 	LbPolicyHost0Port                  int
 	LbPolicyHost1Port                  int