20543022641

Committed 27 Dec 2025 06:53PM UTC coverage: 26.309% (-2.1%) from 28.375%

Build # 20543022641

Build Type

Pull #8

github

Committed by

stillya

Commit Message

polish commit

Pull Request Pull Request #8: feat(agent): made seperation on control plane and cli

Run Details

271 of 938 new or added lines in 7 files covered. (28.89%)

412 of 1566 relevant lines covered (26.31%)

0.27 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.75

/pkg/api/server.go

package api

import (
        "bufio"
        "context"
        "encoding/json"
        "fmt"
        "net"
        "os"
        "path/filepath"
        "sync"
        "time"

        log "log/slog"

        "github.com/pkg/errors"
)

const (
        DefaultSocketPath = "/var/run/wg-relay/control.sock"
        SocketDirMode     = 0755
        SocketFileMode    = 0600
)

//go:generate moq -out controlhandler_mock.go -fmt goimports . ControlHandler
type ControlHandler interface {
        HandleEnable(ctx context.Context, args EnableArgs) error
        HandleDisable(ctx context.Context) error
        HandleReload(ctx context.Context, args ReloadArgs) error
        GetStatus(ctx context.Context) (*StatusResponse, error)
        GetStats(ctx context.Context) (*StatsResponse, error)
}

type Server struct {
        socketPath string
        handler    ControlHandler
        listener   net.Listener
        mu         sync.Mutex
        running    bool
}

func NewServer(socketPath string, handler ControlHandler) (*Server, error) {
        if socketPath == "" {
                socketPath = DefaultSocketPath
        }

        return &Server{
                socketPath: socketPath,
                handler:    handler,
        }, nil
}

func (s *Server) Start(ctx context.Context) error {
        s.mu.Lock()
        if s.running {
                s.mu.Unlock()
                return errors.New("server already running")
        }
        s.mu.Unlock()

        if err := os.RemoveAll(s.socketPath); err != nil {
                return errors.Wrap(err, "failed to remove existing socket")
        }

        socketDir := filepath.Dir(s.socketPath)
        if err := os.MkdirAll(socketDir, SocketDirMode); err != nil {
                return errors.Wrap(err, "failed to create socket directory")
        }

        listener, err := net.Listen("unix", s.socketPath)
        if err != nil {
                return errors.Wrap(err, "failed to listen on unix socket")
        }

        if err := os.Chmod(s.socketPath, SocketFileMode); err != nil {
                listener.Close()
                return errors.Wrap(err, "failed to set socket permissions")
        }

        s.mu.Lock()
        s.listener = listener
        s.running = true
        s.mu.Unlock()

        log.Info("Control API server started", "socket", s.socketPath)

        go s.acceptLoop(ctx)

        return nil
}

func (s *Server) Stop() error {
        s.mu.Lock()
        defer s.mu.Unlock()

        if !s.running {
                return nil
        }

        s.running = false

        if s.listener != nil {
                s.listener.Close()
        }

        os.RemoveAll(s.socketPath)

        log.Info("Control API server stopped")

        return nil
}

func (s *Server) acceptLoop(ctx context.Context) {
        for {
                conn, err := s.listener.Accept()
                if err != nil {
                        s.mu.Lock()
                        running := s.running
                        s.mu.Unlock()

                        if !running {
                                return
                        }

                        log.Error("Accept error", "error", err)
                        continue
                }

                go s.handleConnection(ctx, conn)
        }
}

func (s *Server) handleConnection(ctx context.Context, conn net.Conn) {
        defer conn.Close()

        conn.SetDeadline(time.Now().Add(30 * time.Second))

        scanner := bufio.NewScanner(conn)
        if !scanner.Scan() {
                if err := scanner.Err(); err != nil {
                        log.Error("Failed to read request", "error", err)
                }
                return
        }

        var req Request
        if err := json.Unmarshal(scanner.Bytes(), &req); err != nil {
                s.sendResponse(conn, Response{
                        Success: false,
                        Error:   fmt.Sprintf("invalid request: %v", err),
                })
                return
        }

        resp := s.handleRequest(ctx, &req)
        s.sendResponse(conn, resp)
}

func (s *Server) handleRequest(ctx context.Context, req *Request) Response {
        log.Info("Received command", "command", req.Command, "config_path", req.ConfigPath)

        switch req.Command {
        case CommandEnable:
                if err := s.handler.HandleEnable(ctx, EnableArgs{ConfigPath: req.ConfigPath}); err != nil {
                        return Response{
                                Success: false,
                                Error:   err.Error(),
                        }
                }
                return Response{Success: true}

        case CommandDisable:
                if err := s.handler.HandleDisable(ctx); err != nil {
                        return Response{
                                Success: false,
                                Error:   err.Error(),
                        }
                }
                return Response{Success: true}

        case CommandReload:
                if err := s.handler.HandleReload(ctx, ReloadArgs{ConfigPath: req.ConfigPath}); err != nil {
                        return Response{
                                Success: false,
                                Error:   err.Error(),
                        }
                }
                return Response{Success: true}

        case CommandStatus:
                status, err := s.handler.GetStatus(ctx)
                if err != nil {
                        return Response{
                                Success: false,
                                Error:   err.Error(),
                        }
                }
                return Response{
                        Success: true,
                        Data:    status,
                }

        case CommandStats:
                stats, err := s.handler.GetStats(ctx)
                if err != nil {
                        return Response{
                                Success: false,
                                Error:   err.Error(),
                        }
                }
                return Response{
                        Success: true,
                        Data:    stats,
                }

        default:
                return Response{
                        Success: false,
                        Error:   fmt.Sprintf("unknown command: %s", req.Command),
                }
        }
}

func (s *Server) sendResponse(conn net.Conn, resp Response) {
        data, err := json.Marshal(resp)
        if err != nil {
                log.Error("Failed to marshal response", "error", err)
                return
        }

        data = append(data, '\n')
        if _, err := conn.Write(data); err != nil {
                log.Error("Failed to write response", "error", err)
        }
}

1	package api
2
3	import (
4	"bufio"
5	"context"
6	"encoding/json"
7	"fmt"
8	"net"
9	"os"
10	"path/filepath"
11	"sync"
12	"time"
13
14	log "log/slog"
15
16	"github.com/pkg/errors"
17	)
18
19	const (
20	DefaultSocketPath = "/var/run/wg-relay/control.sock"
21	SocketDirMode = 0755
22	SocketFileMode = 0600
23	)
24
25	//go:generate moq -out controlhandler_mock.go -fmt goimports . ControlHandler
26	type ControlHandler interface {
27	HandleEnable(ctx context.Context, args EnableArgs) error
28	HandleDisable(ctx context.Context) error
29	HandleReload(ctx context.Context, args ReloadArgs) error
30	GetStatus(ctx context.Context) (*StatusResponse, error)
31	GetStats(ctx context.Context) (*StatsResponse, error)
32	}
33
34	type Server struct {
35	socketPath string
36	handler ControlHandler
37	listener net.Listener
38	mu sync.Mutex
39	running bool
40	}
41
42	func NewServer(socketPath string, handler ControlHandler) (*Server, error) {	1✔
43	if socketPath == "" {	1✔
NEW 44	socketPath = DefaultSocketPath	×
NEW 45	}	×
46
47	return &Server{	1✔
48	socketPath: socketPath,	1✔
49	handler: handler,	1✔
50	}, nil	1✔
51	}
52
53	func (s *Server) Start(ctx context.Context) error {	1✔
54	s.mu.Lock()	1✔
55	if s.running {	1✔
NEW 56	s.mu.Unlock()	×
NEW 57	return errors.New("server already running")	×
NEW 58	}	×
59	s.mu.Unlock()	1✔
60		1✔
61	if err := os.RemoveAll(s.socketPath); err != nil {	1✔
NEW 62	return errors.Wrap(err, "failed to remove existing socket")	×
NEW 63	}	×
64
65	socketDir := filepath.Dir(s.socketPath)	1✔
66	if err := os.MkdirAll(socketDir, SocketDirMode); err != nil {	1✔
NEW 67	return errors.Wrap(err, "failed to create socket directory")	×
NEW 68	}	×
69
70	listener, err := net.Listen("unix", s.socketPath)	1✔
71	if err != nil {	1✔
NEW 72	return errors.Wrap(err, "failed to listen on unix socket")	×
NEW 73	}	×
74
75	if err := os.Chmod(s.socketPath, SocketFileMode); err != nil {	1✔
NEW 76	listener.Close()	×
NEW 77	return errors.Wrap(err, "failed to set socket permissions")	×
NEW 78	}	×
79
80	s.mu.Lock()	1✔
81	s.listener = listener	1✔
82	s.running = true	1✔
83	s.mu.Unlock()	1✔
84		1✔
85	log.Info("Control API server started", "socket", s.socketPath)	1✔
86		1✔
87	go s.acceptLoop(ctx)	1✔
88		1✔
89	return nil	1✔
90	}
91
92	func (s *Server) Stop() error {	1✔
93	s.mu.Lock()	1✔
94	defer s.mu.Unlock()	1✔
95		1✔
96	if !s.running {	1✔
NEW 97	return nil	×
NEW 98	}	×
99
100	s.running = false	1✔
101		1✔
102	if s.listener != nil {	2✔
103	s.listener.Close()	1✔
104	}	1✔
105
106	os.RemoveAll(s.socketPath)	1✔
107		1✔
108	log.Info("Control API server stopped")	1✔
109		1✔
110	return nil	1✔
111	}
112
113	func (s *Server) acceptLoop(ctx context.Context) {	1✔
114	for {	2✔
115	conn, err := s.listener.Accept()	1✔
116	if err != nil {	2✔
117	s.mu.Lock()	1✔
118	running := s.running	1✔
119	s.mu.Unlock()	1✔
120		1✔
121	if !running {	2✔
122	return	1✔
123	}	1✔
124
NEW 125	log.Error("Accept error", "error", err)	×
NEW 126	continue	×
127	}
128
129	go s.handleConnection(ctx, conn)	1✔
130	}
131	}
132
133	func (s *Server) handleConnection(ctx context.Context, conn net.Conn) {	1✔
134	defer conn.Close()	1✔
135		1✔
136	conn.SetDeadline(time.Now().Add(30 * time.Second))	1✔
137		1✔
138	scanner := bufio.NewScanner(conn)	1✔
139	if !scanner.Scan() {	1✔
NEW 140	if err := scanner.Err(); err != nil {	×
NEW 141	log.Error("Failed to read request", "error", err)	×
NEW 142	}	×
NEW 143	return	×
144	}
145
146	var req Request	1✔
147	if err := json.Unmarshal(scanner.Bytes(), &req); err != nil {	1✔
NEW 148	s.sendResponse(conn, Response{	×
NEW 149	Success: false,	×
NEW 150	Error: fmt.Sprintf("invalid request: %v", err),	×
NEW 151	})	×
NEW 152	return	×
NEW 153	}	×
154
155	resp := s.handleRequest(ctx, &req)	1✔
156	s.sendResponse(conn, resp)	1✔
157	}
158
159	func (s Server) handleRequest(ctx context.Context, req Request) Response {	1✔
160	log.Info("Received command", "command", req.Command, "config_path", req.ConfigPath)	1✔
161		1✔
162	switch req.Command {	1✔
163	case CommandEnable:	1✔
164	if err := s.handler.HandleEnable(ctx, EnableArgs{ConfigPath: req.ConfigPath}); err != nil {	1✔
NEW 165	return Response{	×
NEW 166	Success: false,	×
NEW 167	Error: err.Error(),	×
NEW 168	}	×
NEW 169	}	×
170	return Response{Success: true}	1✔
171
172	case CommandDisable:	1✔
173	if err := s.handler.HandleDisable(ctx); err != nil {	1✔
NEW 174	return Response{	×
NEW 175	Success: false,	×
NEW 176	Error: err.Error(),	×
NEW 177	}	×
NEW 178	}	×
179	return Response{Success: true}	1✔
180
181	case CommandReload:	1✔
182	if err := s.handler.HandleReload(ctx, ReloadArgs{ConfigPath: req.ConfigPath}); err != nil {	1✔
NEW 183	return Response{	×
NEW 184	Success: false,	×
NEW 185	Error: err.Error(),	×
NEW 186	}	×
NEW 187	}	×
188	return Response{Success: true}	1✔
189
190	case CommandStatus:	1✔
191	status, err := s.handler.GetStatus(ctx)	1✔
192	if err != nil {	1✔
NEW 193	return Response{	×
NEW 194	Success: false,	×
NEW 195	Error: err.Error(),	×
NEW 196	}	×
NEW 197	}	×
198	return Response{	1✔
199	Success: true,	1✔
200	Data: status,	1✔
201	}	1✔
202
203	case CommandStats:	1✔
204	stats, err := s.handler.GetStats(ctx)	1✔
205	if err != nil {	1✔
NEW 206	return Response{	×
NEW 207	Success: false,	×
NEW 208	Error: err.Error(),	×
NEW 209	}	×
NEW 210	}	×
211	return Response{	1✔
212	Success: true,	1✔
213	Data: stats,	1✔
214	}	1✔
215
216	default:	1✔
217	return Response{	1✔
218	Success: false,	1✔
219	Error: fmt.Sprintf("unknown command: %s", req.Command),	1✔
220	}	1✔
221	}
222	}
223
224	func (s *Server) sendResponse(conn net.Conn, resp Response) {	1✔
225	data, err := json.Marshal(resp)	1✔
226	if err != nil {	1✔
NEW 227	log.Error("Failed to marshal response", "error", err)	×
NEW 228	return	×
NEW 229	}	×
230
231	data = append(data, '\n')	1✔
232	if _, err := conn.Write(data); err != nil {	1✔
NEW 233	log.Error("Failed to write response", "error", err)	×
NEW 234	}	×
235	}

stillya / wg-relay / 20543022641

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous