Agent graceful shutdown

[LaurenceJJones]

Agent graceful shutdown

[negasus]

Thanks. Feel free for PR )

[LaurenceJJones]

Further testing shows because there is no shutdown functionality if you are running as a service and you issue a SIGINT to shutdown the service cant breaks the loops so systemd will issue a SIGKILL in the end which we should improve

[LaurenceJJones]

Hey 👋🏻 So I checked it out and my lack of experience on how SPOP protocol works really has me guessing a few things so I probably wont push any changes but if anyone happens to stumble on this issue here a implementation which allows the shutdown the server but is not graceful due to fact that internally within the frames we dont check if any frames are being processed:

// You can edit this code!
// Click here and start typing.
package main

import (
	"context"
	"errors"
	"log"
	"math/rand"
	"net"
	"os"
	"os/signal"
	"syscall"

	"github.com/negasus/haproxy-spoe-go/action"
	"github.com/negasus/haproxy-spoe-go/agent"
	"github.com/negasus/haproxy-spoe-go/logger"
	"github.com/negasus/haproxy-spoe-go/request"
)

func HandleSignals(ctx context.Context) error {
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGTERM, os.Interrupt)

	select {
	case s := <-signalChan:
		switch s {
		case syscall.SIGTERM:
			return errors.New("received SIGTERM")
		case os.Interrupt: // cross-platform SIGINT
			return errors.New("received interrupt")
		}
	case <-ctx.Done():
		return ctx.Err()
	}

	return nil
}

func main() {
	g, ctx := errgroup.WithContext(context.Background())

	g.Go(func() error {
		return HandleSignals(ctx)
	})

	g.Go(func() error {
		errorChan := make(chan error, 1)
		defer close(errorChan)

		log.Print("listen 3000")

		listener, err := net.Listen("tcp4", "127.0.0.1:3000")
		if err != nil {
			log.Printf("error create listener, %v", err)
			return err
		}
		defer listener.Close()
	
		go func() {
			a := agent.New(handler, logger.NewDefaultLog())
			if err := a.Serve(listener); err != nil {
				errorChan <- err
			}
		}()
		select {
		case err := <-errorChan:
			return err
		case <-ctx.Done():
			return nil
		}
	})

	if err := g.Wait(); err != nil {
		//Handle exit codes here
		os.Exit(1)
	}
}

func handler(req *request.Request) {

	log.Printf("handle request EngineID: '%s', StreamID: '%d', FrameID: '%d' with %d messages\n", req.EngineID, req.StreamID, req.FrameID, req.Messages.Len())

	messageName := "get-ip-reputation"

	mes, err := req.Messages.GetByName(messageName)
	if err != nil {
		log.Printf("message %s not found: %v", messageName, err)
		return
	}

	ipValue, ok := mes.KV.Get("ip")
	if !ok {
		log.Printf("var 'ip' not found in message")
		return
	}

	ip, ok := ipValue.(net.IP)
	if !ok {
		log.Printf("var 'ip' has wrong type. expect IP addr")
		return
	}

	ipScore := rand.Intn(100)

	log.Printf("IP: %s, send score '%d'", ip.String(), ipScore)

	req.Actions.SetVar(action.ScopeSession, "ip_score", ipScore)
}

Maybe we can push to readme.md but up to you ✨