Add handler outcome Prometheus metric for SLO-based kill rate alerting

[milos-lk]

We currently monitor egress handler kills via a logs that counts killed vs validated log lines. This is noisy at low traffic because small denominators cause the ratio to spike, and it mixes request-time events (validation) with process-exit events (kills), which skews the ratio for long-running egresses.

This PR adds a single prometheus counter, livekit_egress_handler_results_total{type, result}, that tracks terminal handler process outcomes. Every handler that runs records exactly one result when it exits: completed, process_error, aborted, killed_cpu, killed_oom, or killed_shutdown. The type label carries the egress request type for drill-down.

To correctly attribute kill reasons at process exit time, the ProcessManager now maintains a killReasons map. When the monitor triggers a kill (CPU or OOM), it stores the reason before sending the kill signal. When cmd.Wait() returns in the goroutine, it consumes the stored reason via GetKillReason and records the appropriate metric. The consume-on-read design prevents double-counting when ProcessFinished cleans up.

The KillProcess method now takes an explicit reason parameter instead of deriving the reason from error string matching, which keeps the metric labels type-safe and co-located with the constants they reference.

[biglittlebigben]

In this code path, it looks like failing to call GetKillReason will cause a leak? This feels a bit fragile. Is it possible to refactor the Abort/createErr flow in a way where the activeHanders entry get removed after we read the kill reason (at what point we should be able to stash killReason in Process). Alternatively is there a way to register a "UpdateKillReason" callback that sets the prometheus value (either sync or in a goroutine) and gets called at the point where we are currently setting the killReason entry?

[milos-lk]

In this code path, it looks like failing to call GetKillReason will cause a leak?

only if ProcessFinished isn't called but if that happens we are in a deeper kimchi and leaking active handlers map and a lot more of other things. But could probably be made simpler (without the need for a map) if we delay removal from active handlers after the reason is read - will update that.

[biglittlebigben]

The AbortProcess path doesn't call ProcessFinished. That's why we currently remove the Process entry from activeHandler in AbortProcess. That the path I think is dangerous.

[milos-lk]

hm - not sure if this is the case - as far as i can see AbortProcess() is called from StartEgress only after handler is launched at which point it needs to return from cmd.Wait() - and invoke processEnded which calls processFinished - or I missed something?

[biglittlebigben]

Right, today, if we call AbortProcess, launchErr is nil and thus we call processEnded after cmd.Wait. I was focusing on the StartEgress implementation only. It is all very fragile though. This also makes me wonder why AbortEgress removes the Provess from activeHandlers.

[milos-lk]

It was introduced in https://github.com/livekit/egress/pull/769/changes
looking in the code there - cmd.wait would have been executed - removing from the list of active handlers would have been done in process finished as well - so not seeing obvious reason to for immediate removal. Maybe it was defensive. Even if we have some other call in a meanwhile the worst thing that could happen is failing IPC but we handle that gracefully as far as I can see. Maybe @frostbyte73 remembers?

[milos-lk]

Delayed removal has also one nice side effect - consider this scenario:

  ---                                                                                     
  Goroutine 2 (spawned at end of launchProcess):                                          
  cmd.Wait()  ← blocking, waiting for handler process to exit                             
                                                                                          
  Goroutine 1 (RPC handler, continuing StartCloudEgress):                                 
  createErr := <-errChan  ← IO create failed                                              
  → enters case createErr != nil            
  → AbortProcess(egressID, createErr)                                                     
                                                                                        
  Inside AbortProcess:                                                                    
  pm.mu.Lock()                                                                          
  h.killReason = stats.ResultAborted   ← reason stored on Process                         
  pm.mu.Unlock()                       ← lock released                                  
                                                                                          
  h.kill(err)                          ← tries gRPC KillEgress over Unix socket           
                                                                                          
  Two sub-scenarios from here:                                                            
                                                                                          
  ---
  A) Handler is still alive:                                                              
  - KillEgress gRPC succeeds → handler shuts down → process exits                       
  - h.ipcHandlerClient.Close() closes the connection             
  - Goroutine 2: cmd.Wait() returns → GetKillReason reads "aborted" from h.killReason →   
  records metric → processEnded → ProcessFinished closes IPC (already closed, no-op),     
  deletes from activeHandlers                                                             
  - Clean.                                                                                
                                                                                          
  B) Handler is already dead (crashed before abort):                                    
  - h.kill() enters p.closed.Once() → tries KillEgress gRPC → socket is stale → blocks    
  - h.ipcHandlerClient.Close() hasn't been reached yet (goroutine 1 stuck on line above)  
  - Goroutine 2: cmd.Wait() returns immediately (process already dead)                    
  - GetKillReason → pm.mu.RLock() → succeeds (goroutine 1 doesn't hold the lock) → reads  
  "aborted" → records metric                                                            
  - processEnded → ProcessFinished → pm.mu.Lock() → succeeds → finds handler in map →     
  calls p.ipcHandlerClient.Close() → this closes the underlying connection → unblocks the 
  stuck gRPC call in goroutine 1                                                          
  - ProcessFinished deletes from activeHandlers, done                                   
  - Goroutine 1: h.kill() returns (unblocked by the close) → h.ipcHandlerClient.Close()   
  again → no-op, already closed                                                         
  - Both goroutines complete. No leak. 

So scenario B fixes the goroutine leak as far as I can see

[milos-lk]

Basically with this change, the handler stays in activeHandlers, so ProcessFinished finds it, calls Close(), and unblocks the stuck kill(). I roped in claude to simulate stuck process just to make sure Close() unblocks IPC which it does:

package ipc

import (
	"context"
	"net"
	"os"
	"path"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/emptypb"
)

type blockingHandler struct {
	UnimplementedEgressHandlerServer
	blocked chan struct{}
}

func (h *blockingHandler) KillEgress(_ context.Context, _ *KillEgressRequest) (*emptypb.Empty, error) {
	<-h.blocked
	return &emptypb.Empty{}, nil
}

func TestCloseUnblocksStuckKillOnStaleSocket(t *testing.T) {
	tmpDir, err := os.MkdirTemp("/tmp", "ipc-test-")
	require.NoError(t, err)
	defer os.RemoveAll(tmpDir)

	// Start a server and create a client, then stop the server to simulate a dead handler
	srv := grpc.NewServer()
	RegisterEgressHandlerServer(srv, &blockingHandler{blocked: make(chan struct{})})
	require.NoError(t, StartHandlerListener(srv, tmpDir))

	client, err := NewHandlerClient(tmpDir)
	require.NoError(t, err)

	srv.GracefulStop()

	// Re-create the socket file without serving — simulates stale socket from dead process
	_, err = net.Listen("unix", path.Join(tmpDir, handlerAddress))
	require.NoError(t, err)

	// KillEgress blocks on the stale socket
	killReturned := make(chan error, 1)
	go func() {
		_, err := client.KillEgress(context.Background(), &KillEgressRequest{Error: "test"})
		killReturned <- err
	}()

	select {
	case <-killReturned:
		t.Fatal("KillEgress should be stuck on stale socket")
	case <-time.After(2 * time.Second):
	}

	// Close unblocks it
	require.NoError(t, client.Close())

	select {
	case err := <-killReturned:
		require.Error(t, err)
	case <-time.After(5 * time.Second):
		t.Fatal("Close() did not unblock the stuck KillEgress call")
	}
}

[github-actions]

AV-sync stats summary: view in run #26257363168