client.go

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/http/httptrace"
	"time"

	"github.com/hashicorp/go-retryablehttp"
	"github.com/replicate/go/httpclient"
)

const SingleRequestTimeout = 5 * time.Second

func main() {
	client := NewForwarderHTTPClient()

	ctx := context.Background()

	trace := &httptrace.ClientTrace{
		GotConn: func(connInfo httptrace.GotConnInfo) {
			fmt.Printf("got conn: %+v\n", connInfo)
		},
	}

	for {
		fmt.Printf("making a request\n")

		req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8088", nil)

		req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace))

		resp, err := client.Do(req)
		if err != nil {
			fmt.Printf("request error: %s\n", err.Error())
			continue
		}
		if _, err := io.Copy(io.Discard, resp.Body); err != nil {
			log.Fatal(err)
		}
		resp.Body.Close()
	}

}

func NewForwarderHTTPClient() *http.Client {
	retryClient := &retryablehttp.Client{
		HTTPClient:   httpclient.DefaultPooledClient(),
		Logger:       nil, // "logging" is provided by OTel transport on the web client
		RetryWaitMin: 100 * time.Millisecond,
		RetryWaitMax: 2 * time.Second,
		RetryMax:     2,
		CheckRetry:   ForwarderRetryPolicy,
		Backoff:      retryablehttp.DefaultBackoff,
	}
	retryClient.HTTPClient.Timeout = SingleRequestTimeout
	return retryClient.StandardClient()
}

func ForwarderRetryPolicy(ctx context.Context, resp *http.Response, err error) (bool, error) {
	// do not retry on context.Canceled or context.DeadlineExceeded
	if ctx.Err() != nil {
		return false, ctx.Err()
	}

	// Return 429s to the client without retrying, so as not to consume any more
	// than one "token" from their token bucket at the remote end.
	if err == nil && resp != nil && resp.StatusCode == http.StatusTooManyRequests {
		return false, nil
	}

	return retryablehttp.DefaultRetryPolicy(ctx, resp, err)
}

go.mod

module blah.com/v1

go 1.20

require (
	github.com/hashicorp/go-retryablehttp v0.7.4
	github.com/replicate/go v0.0.0-20231102192516-076a12128352
)

require (
	github.com/felixge/httpsnoop v1.0.3 // indirect
	github.com/go-logr/logr v1.2.4 // indirect
	github.com/go-logr/stdr v1.2.2 // indirect
	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.45.0 // indirect
	go.opentelemetry.io/otel v1.19.0 // indirect
	go.opentelemetry.io/otel/metric v1.19.0 // indirect
	go.opentelemetry.io/otel/trace v1.19.0 // indirect
)

go.sum

github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
github.com/felixge/httpsnoop v1.0.3/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.2.4 h1:g01GSCwiDw2xSZfjJ2/T9M+S6pFdcNtFYsp+Y43HYDQ=
github.com/go-logr/logr v1.2.4/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=
github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48=
github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
github.com/hashicorp/go-retryablehttp v0.7.4 h1:ZQgVdpTdAL7WpMIwLzCfbalOcSUdkDZnpUv3/+BxzFA=
github.com/hashicorp/go-retryablehttp v0.7.4/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/replicate/go v0.0.0-20231102192516-076a12128352 h1:dUfmwaG9WqnJ7TkEZfDtTvIVgwE9zUPBLW8nifW/U1Y=
github.com/replicate/go v0.0.0-20231102192516-076a12128352/go.mod h1:GyvRfBYP2DdTSKuodGgxr+qTUN0rmTlYaVfoXxXQ2nc=
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.45.0 h1:x8Z78aZx8cOF0+Kkazoc7lwUNMGy0LrzEMxTm4BbTxg=
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.45.0/go.mod h1:62CPTSry9QZtOaSsE3tOzhx6LzDhHnXJ6xHeMNNiM6Q=
go.opentelemetry.io/otel v1.19.0 h1:MuS/TNf4/j4IXsZuJegVzI1cwut7Qc00344rgH7p8bs=
go.opentelemetry.io/otel v1.19.0/go.mod h1:i0QyjOq3UPoTzff0PJB2N66fb4S0+rSbSB15/oyH9fY=
go.opentelemetry.io/otel/metric v1.19.0 h1:aTzpGtV0ar9wlV4Sna9sdJyII5jTVJEvKETPiOKwvpE=
go.opentelemetry.io/otel/metric v1.19.0/go.mod h1:L5rUsV9kM1IxCj1MmSdS+JQAcVm319EUrDVLrt7jqt8=
go.opentelemetry.io/otel/trace v1.19.0 h1:DFVQmlVbfVeOuBRrwdtaehRrWiL1JoVs9CPIQ1Dzxpg=
go.opentelemetry.io/otel/trace v1.19.0/go.mod h1:mfaSyvGyEJEI0nyV2I4qhNQnbBOUUmYZpYojqMnX2vo=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

server.go

package main

import (
	"bufio"
	"fmt"
	"io"
	"log"
	"net"
)

// main serves as the program entry point
func main() {
	// create a tcp listener on the given port
	listener, err := net.Listen("tcp", "127.0.0.1:8088")
	if err != nil {
		log.Fatalln("failed to create listener, err:", err)
	}
	fmt.Printf("listening on %s\n", listener.Addr())

	// listen for new connections
	for {
		conn, err := listener.Accept()
		if err != nil {
			fmt.Println("failed to accept connection, err:", err)
			continue
		}

		fmt.Printf("accepted connection from %s\n", conn.RemoteAddr())

		go handleConnection(conn)
	}
}

// handleConnection handles the lifetime of a connection
func handleConnection(conn net.Conn) {
	defer conn.Close()
	reader := bufio.NewReader(conn)
	for {
		// read client request data
		bytes, err := reader.ReadBytes(byte('\n'))
		if err != nil {
			if err != io.EOF {
				fmt.Println("failed to read data, err:", err)
			}
			fmt.Printf("connection closed by client: %s\n", conn.RemoteAddr())
			return
		}
		fmt.Printf("request: %s", bytes)

		// Ignore the request! This is what we think is happening at CoreWeave
		// if string(bytes) == "\r\n" {
		// 	fmt.Fprintf(conn, "HTTP/1.1 200 OK\r\n")
		// 	fmt.Fprintf(conn, "Content-Length: 12\r\n")
		// 	fmt.Fprintf(conn, "Content-Type: text/plain\r\n")
		// 	fmt.Fprintf(conn, "\r\n")
		// 	fmt.Fprintf(conn, "Hello World!")
		// }
	}
}