criteo / haproxy-spoe-go

An implementation of the SPOP protocol in Go. https://www.haproxy.org/download/2.0/doc/SPOE.txt

Please note: This repository is currently unmaintained. Please switch to project https://github.com/negasus/haproxy-spoe-go if you want to use a golang version of haproxy-spoe library.

SPOE in Go

An implementation of the SPOP protocol in Go. (https://www.haproxy.org/download/2.0/doc/SPOE.txt)

SPOE

From Haproxy's documentation :

SPOE is a feature introduced in HAProxy 1.7. It makes possible the communication with external components to retrieve some info. The idea started with the problems caused by most ldap libs not working fine in event-driven systems (often at least the connect() is blocking). So, it is hard to properly implement Single Sign On solution (SSO) in HAProxy. The SPOE will ease this kind of processing, or we hope so.

Now, the aim of SPOE is to allow any kind of offloading on the streams. First releases, besides being experimental, won't do lot of things. As we will see, there are few handled events and even less actions supported. Actually, for now, the SPOE can offload the processing before "tcp-request content", "tcp-response content", "http-request" and "http-response" rules. And it only supports variables definition. But, in spite of these limited features, we can easily imagine to implement SSO solution, ip reputation or ip geolocation services.

How to use

package main

import (
	"fmt"
	"net"

	"log"

	spoe "github.com/criteo/haproxy-spoe-go"
)

func getReputation(ip net.IP) (float64, error) {
	// implement IP reputation code here
	return 1.0, nil
}

func main() {
	agent := spoe.New(func(messages *spoe.MessageIterator) ([]spoe.Action, error) {
		reputation := 0.0

		for messages.Next() {
			msg := messages.Message

			if msg.Name != "ip-rep" {
				continue
			}

			var ip net.IP
			for msg.Args.Next() {
				arg := msg.Args.Arg

				if arg.Name == "ip" {
					var ok bool
					ip, ok = arg.Value.(net.IP)
					if !ok {
						return nil, fmt.Errorf("spoe handler: expected ip in message, got %+v", ip)
					}
				}
			}

			var err error
			reputation, err = getReputation(ip)
			if err != nil {
				return nil, fmt.Errorf("spoe handler: error processing request: %s", err)
			}
		}

		return []spoe.Action{
			spoe.ActionSetVar{
				Name:  "reputation",
				Scope: spoe.VarScopeSession,
				Value: reputation,
			},
		}, nil
	})

	if err := agent.ListenAndServe(":9000"); err != nil {
		log.Fatal(err)
	}
}

haproxy-spoe-go's Issues

Memory/Goroutine leak

Summary

When HAProxy sends diconnect frame with specific "status codes" (e.g. I/O error), we leak goroutines. This leads to memory leak. Eventually, the process dies to OOM.

In our test environment, we observed that if there is a high rate of such disconnect frames sent by HAPRoxy, the memory consumption can reach upto 120G in less than 10 minutes. Note: We were sending several KBs of data in each haproxy --> SPOA call; at 100k requests/second.

Details

Here is my analysis. All the details below may not be correct - but overall it does look like there is a leak in "haproxy-spoe-go".

All the goroutines seem to be stuck here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/notify.go#L130

Here is the reason for goroutine leak:

This is the "main run loop": https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L26

There are two main loops (conceptually):

1. "response loop" to send the response back to haproxy from SPOA. This is done as part of a gouroutine here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L106. Two things to note here:

   • This go routine will read data from channel "frames" and send data to haproxy.
   • This goroutine will exit when there is an event on channel "done". Once it exits, there is no one to read from channel "frames". Hence, any writes on channel "frames" will block.

2. "request loop" to read data sent by haproxy and act on it. This is not a separate go routine, rather part of the "main run loop" here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L122

Here is what happens.

1. Due to short processing timeout of 10ms on haproxy side for SPOA calls, there are a lot of disconnects sent by haproxy to SPOA.
2. When SPOA rcvs a disconnect frame from haproxy, it calls method "handleDisconnect" here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L140. "handleDisconnect" returns a non-nil error, because the "status-code" is set to "I/O error" in the disconnect frame sent by haproxy to SPOA - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/disconnect.go#L64. (We can look at haproxy code to understand when haproxy sets this status code.).
3. As can be seen here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L142 - on a non-nil error while handling disconnect, the "main run loop" will return.
4. When the "main run loop" returns, channel "done" is closed - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L29
5. When channel "done" is closed, the "response loop" goroutine will exit.
6. Since there is no one to read from "frames" channel, all the gouroutines launched as part of "runWorker" here - https://github.com/criteo/haproxy-spoe-go/blob/v1.0.6/conn.go#L136 - will be blocked on the frames channel... INDEFINITELY
7. AND WE HAVE A GOROTUINE LEAK... hence a memory leak.

Misc

Memory leak

Error message from logs

Go routine leak

Attached are some snapshots of the logs messages from our golang process and pprof profiles.

panic: sync: negative WaitGroup counter

Hello,

While doing some stress tests here, we've found the following error:

panic: sync: negative WaitGroup counter

Here's the stack trace:

panic: sync: negative WaitGroup counter

goroutine 2650 [running]:
sync.(*WaitGroup).Add(0xc001822f80, 0xffffffffffffffff)
        /usr/local/go/src/sync/waitgroup.go:74 +0x147
sync.(*WaitGroup).Done(0xc001822f80)
        /usr/local/go/src/sync/waitgroup.go:99 +0x34
github.com/criteo/haproxy-spoe-go.(*conn).run(0xc00188b220, 0xc00002a600, 0x0, 0x0)
        /go/pkg/mod/github.com/criteo/[email protected]/conn.go:131 +0xa00
github.com/criteo/haproxy-spoe-go.(*Agent).Serve.func1(0x97c100, 0xc00019c8b8, 0xc00002a600)
        /go/pkg/mod/github.com/criteo/[email protected]/spoe.go:100 +0xcc
created by github.com/criteo/haproxy-spoe-go.(*Agent).Serve
        /go/pkg/mod/github.com/criteo/[email protected]/spoe.go:93 +0x167

Thanks :D

question: decoding headers

I'm very new to both Go and HAProxy, but I've got a good working prototype going using your code; thanks!

The last issue I have is how to decode headers. I'm sending the headers from HAproxy using req.hdrs_bin. I've dumped the data and did a little decoding (in Elixir; but it works). I confirmed the format: an integer indicating the length of bytes to come, the bytes, and the EOF marker is double NUL chars.

[
  <<4>>, "h", "o", "s", "t",
  <<14>>, "l", "o", "c", "a", "l", "h", "o", "s", "t", ":", "8", "0", "5", "5"
  <<10>>, "u", "s", "e", "r", "-", "a", "g", "e", "n", "t",
  <<11>>, "c", "u", "r", "l", "/", "7", ".", "5", "4", ".", "0",
  <<6>>, "a", "c", "c", "e", "p", "t",
  <<16>>, "a", "p", "p", "l", "i", "c", "a", "t", "i", "o", "n", "/", "j", "s", "o", "n",
  <<0>>, <<0>>
]

I'm searching through the code seeing if there's a decoder for this built in. I found decodeKVs but that's not exported and reading it, I don't think it would decode this.

Have you solved the header problem? If you could provide a code snippet to do so, that would be great. I would be happy to open a PR to update the README to include that.

Edit

I've used my 1 day old Go skills to come up with this

// Parses the HAProxy binary format into a list of strings. The byte format is:
//   <chunk size><chunk> ... \0 \0
// For example the string "host" would be encoded as:
//   []byte{4 ,104 ,111 ,115 ,116, 0, 0}
// 4 is the length of the string, then the four bytes. more chunks could follow
// the end of message indicator is the double NIL chars
func decodeStrings(bytes []byte) []string {
	var res []string
	offset := 0

	for offset < len(bytes) {
		size := int(bytes[offset])
		offset += 1

		if size > 0 {
			str := string(bytes[offset : offset+size])
			res = append(res, str)
		}

		offset += size
	}

	return res
}

func decodeHeaders(bytes []byte) map[string]string {
	res := make(map[string]string)
	strings := decodeStrings(bytes)
	offset := 0

	for offset < len(strings) {
		key := strings[offset]
		value := strings[offset+1]
		res[key] = value
		offset += 2
	}

	return res
}

Question : SPOE: [auth-agents] failed to process messages: code=1

Hi

I see this error in my haproxy log whenever I do a CURL request.

Sep 3 03:29:46 172.17.0.8 haproxy[15289]: SPOE: [auth-agents] failed to process messages: code=1

I am not able to enable more debugs to figure out reason for this error.

Is there any way I can see more info on message transactions of this code and haproxy ?

thanks for the help.

Incompatibility with HAProxy's nbthread > 1

When HAProxy is running 1 process and 1 thread, everything works smoothly.
If running multiple processes (nbproc > 1), this would still works smoothly.
That said when I enabled multiple processes, HAProxy does not announce async mode and haproxy-spoe-go lib simply refuses the connection:

INFO[3397] spoe: hello from 127.0.0.1:41537: map[capabilities:pipelining engine-id:6A27BD77-B2EC-475F-B258-E59B68CC3E4E max-frame-size:16380 supported-versions:2.0] 
ERRO[3397] spoe: error handling connection: hello: expected capabilities [async pipelining] 

Would be great if the library can adapt itself to the capabilities announced by HAProxy

too short buffer for encoding in encodeBytes()

It seems there is a bug in the function encodeBytes(): if my action Value string length is longer than the message arg, then the function detects the buffer to store that data is too short and does not encode it.
Here is my use case: I have an Agent which can do DNS resolution and when I do reverse lookup, the host name is usually way longer than the IP address. In such case, I can see that the response frame contains an empty value.
Of course max-frame-size is big enough: 16380 in my case

Is this project still alive?

Hi, I can see many pull-request with absolutely no reaction by any maintainer. Are there any maintainers or is this project abandoned?

missing pipelining capabilities

Hello,

I've been using the library, but got the following error:

time="2020-11-23T17:53:16-03:00" level=debug msg="spoe: hello from 127.0.0.1:53794: map[capabilities: healthcheck:true max-frame-size:16380 supported-versions:2.0]" func="github.com/criteo/haproxy-spoe-go.(*conn).handleHello" file="/home/rkatz/go/pkg/mod/github.com/criteo/[email protected]/hello.go:30"

I'm using HAProxy 2.3, but it fails also with 2.4.

The following is the config I've been using, but apparently even forcing the pipelining capabilities (or trying to change the max-frame-size to make some tests) it doesn't seems to reflect in the hello packet:

[modsecurity]
spoe-agent modsecurity-agent
    max-frame-size 1024
    messages modsecurity
    option var-prefix modsec
    timeout hello 10s
    timeout idle  2m
    timeout processing 1000ms
    use-backend spoe-modsecurity
    log global

spoe-message modsecurity
    args method=method path=path query=query reqver=req.ver ip=src reqhdrs=req.hdrs_bin reqbody=req.body ignorerules=var(txn.ignorerules) srvip=dst srvport=dst_port
    event on-frontend-http-request

Thanks

When SPOE handler returns an error, haproxy-spoe-go does not return anything to HAProxy and the request makes progress in HAProxy only when we hit the timeout for HAProxy <--> SPOA.

SETUP

Here is what I have put in hapoxy.config:

# register spoa
filter spoe engine myspoa config .../myspoa.conf

http-request send-spoe-group myspoa module1

Here is myspoa.conf:

[myspoa]
spoe-agent myspoa
    groups module1
    timeout hello 2s
    timeout idle  2m
    timeout processing 10s # ***** CRUCIAL FOR THIS ISSUE
    use-backend agents
    log global

spoe-group module1
    messages module1

spoe-message module1

In my main.go:

func main() {
	log.Println("Hello from SPOA!")
	logurs.SetLevel(logurs.DebugLevel) // enabled debug log in SPOE library

	// TODO: can specify a custom config for the SPOA e.g. update max connections
	agent := spoe.New(customSpoeHandler)

	if err := agent.ListenAndServe(":12345"); err != nil {
		log.Fatal(err)
	}
}

Let's say in some situations customSpoeHandler returns nil, fmt.Errorf("Some error message"). This could be due to any reason, e.g. a message which we do not want to handle.

What I expect?

SPOA will still return something to HAPRoxy immediately so that the request can make progress in HAProxy.

What happens?

When I send a request using curl, the response comes back after 10 seconds, the timeout configured for HAProxy-SPOA communication.

Possible issue

Most likely due to an unhandled condition here - https://github.com/criteo/haproxy-spoe-go/blob/master/notify.go#L111

	actions, err := c.handler(messages)
	if err != nil {
		return errors.Wrap(err, "handle notify") // TODO return proper response
	}

Workaround

The workaround seems to be that customSpoeHandler should always return nil in the error field. It seems fine to return nil spoe action.

When is the new release?

Last release 1.0.6 was in Jan 2021. I see that there have been several commits since then. These unreleased commits fix some critical issues e.g. memory corruption fix in 003022c

Can you please release a new version of this framework?

Correlate request and response messages

Is it possible to correlate the request and response messages that are received by the agent code? In another go spoe library there was a StreamID that I had access to which seemed to be the same for the request and response messages. It doesn't seem that I have access to that StreamID though. I need something to correlate the request/response so I can generate a database record of the call from a client. Below is my agent config. Thanks again!

[hello-agent]
spoe-agent hello-agent
    messages hello-request hello-response
    option var-prefix hello
    timeout hello 2s
    timeout idle  2m
    timeout processing 10ms
    use-backend hello-agents
    log global

spoe-message hello-request
    args ip=src body=req.body uri=capture.req.uri url=url method=capture.req.method ver=capture.req.ver host=req.hdr(Host)
    event on-frontend-http-request

spoe-message hello-response
    args body=res.body status=status
    event on-http-response

Returning multiple values from the Agent

Is there any way to return multiple values from the Agent to HAProxy ?

For example, can I add more values to the following statement ?

return []spoe.Action{
spoe.ActionSetVar{
Name: "is_authenticated",
Scope: spoe.VarScopeSession,
Value: authenticated,
},
}, nil

Error handling connection - expected pipelining capability

I created a simple hello agent which is doing nothing more than printing out the message name. When I start the agent I am getting the following warn message every second or so:

WARN[0057] spoe: error handling connection: hello: expected pipelining capability

I was able to get the warning to go away by commenting out the line "option spop-check" in my agent backend:

backend hello-agents
mode tcp
balance roundrobin
timeout connect 5s # greater than hello timeout
timeout server 3m # greater than idle timeout
#option spop-check
server agent1 127.0.0.1:9000 check

I tried finding out what the spop-check is for but haven't found any documentation on it. I was just taking a configuration from the haproxy blog. I am currently using the community edition of HAProxy but will be moving to the enterprise edition. Just wondered if you had any insights to this issue. Thanks for your help!

Can I use the SID from haproxy

I use this great module, thanks for writing.
Can I use the stream id from the event?

SPOE: [agent-on-http-req] <EVENT:on-frontend-http-request> sid=88 st=0 0/0/0/0/0 1/1 0/0 10/33
00000058:streams.srvrep[0028:002a]: HTTP/1.0 200 
00000058:streams.srvhdr[0028:002a]: content-type: text/plain
00000058:streams.srvcls[0028:002a]
SPOE: [agent-on-http-res] <EVENT:on-http-response> sid=88 st=0 0/0/0/0/0 1/1 0/0 10/33