diff --git a/apps/iot/src/ctrl/ctrl_acceptor.erl b/apps/iot/src/ctrl/ctrl_acceptor.erl new file mode 100644 index 0000000..cc87282 --- /dev/null +++ b/apps/iot/src/ctrl/ctrl_acceptor.erl @@ -0,0 +1,41 @@ +%%%------------------------------------------------------------------- +%% @doc Control socket acceptor. +%% @end +%%%------------------------------------------------------------------- + +-module(ctrl_acceptor). + +-export([start_link/1, init/1]). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link(gen_tcp:socket()) -> {ok, pid()} | {error, term()}. +start_link(ListenSocket) -> + proc_lib:start_link(?MODULE, init, [ListenSocket]). + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec init(gen_tcp:socket()) -> ok. +init(ListenSocket) -> + ok = iot_log:set_metadata(), + proc_lib:init_ack({ok, self()}), + accept_loop(ListenSocket). + +-spec accept_loop(gen_tcp:socket()) -> ok. +accept_loop(ListenSocket) -> + case gen_tcp:accept(ListenSocket) of + {ok, Socket} -> + {ok, Pid} = ctrl_channel:start(), + ok = gen_tcp:controlling_process(Socket, Pid), + ok = ctrl_channel:set_socket(Pid, Socket), + accept_loop(ListenSocket); + {error, closed} -> + ok; + {error, Reason} -> + logger:error("[ctrl_acceptor] accept failed: ~p", [Reason]), + exit({accept_failed, Reason}) + end. diff --git a/apps/iot/src/ctrl/ctrl_channel.erl b/apps/iot/src/ctrl/ctrl_channel.erl new file mode 100644 index 0000000..6782074 --- /dev/null +++ b/apps/iot/src/ctrl/ctrl_channel.erl @@ -0,0 +1,116 @@ +%%%------------------------------------------------------------------- +%% @doc Single control socket channel. +%% @end +%%%------------------------------------------------------------------- + +-module(ctrl_channel). +-behaviour(gen_server). + +-export([start/0, set_socket/2]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-record(state, { + socket = undefined :: gen_tcp:socket() | undefined +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start() -> {ok, pid()} | ignore | {error, term()}. +start() -> + gen_server:start(?MODULE, [], []). + +-spec set_socket(pid(), gen_tcp:socket()) -> ok. +set_socket(Pid, Socket) -> + gen_server:call(Pid, {set_socket, Socket}). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +-spec init([]) -> {ok, #state{}}. +init([]) -> + ok = iot_log:set_metadata(), + {ok, #state{}}. + +-spec handle_call(term(), {pid(), term()}, #state{}) -> + {reply, term(), #state{}} | {stop, term(), term(), #state{}}. +handle_call({set_socket, Socket}, _From, State = #state{socket = undefined}) -> + case inet:setopts(Socket, [{packet, 2}, {active, once}]) of + ok -> + {reply, ok, State#state{socket = Socket}}; + {error, Reason} -> + logger:warning("[ctrl_channel] failed to activate socket: ~p", [Reason]), + gen_tcp:close(Socket), + {stop, Reason, {error, Reason}, State} + end; +handle_call(_Request, _From, State = #state{}) -> + {reply, ok, State}. + +-spec handle_cast(term(), #state{}) -> {noreply, #state{}}. +handle_cast(_Request, State = #state{}) -> + {noreply, State}. + +-spec handle_info(term(), #state{}) -> {noreply, #state{}} | {stop, term(), #state{}}. +handle_info({tcp, Socket, Data}, State = #state{socket = Socket}) -> + ok = send_response(Socket, dispatch(Data)), + ok = inet:setopts(Socket, [{active, once}]), + {noreply, State}; +handle_info({tcp_closed, Socket}, State = #state{socket = Socket}) -> + {stop, normal, State}; +handle_info({tcp_error, Socket, Reason}, State = #state{socket = Socket}) -> + logger:warning("[ctrl_channel] socket error: ~p", [Reason]), + {stop, Reason, State}; +handle_info(Info, State = #state{}) -> + logger:warning("[ctrl_channel] ignore unknown info: ~p", [Info]), + {noreply, State}. + +-spec terminate(term(), #state{}) -> ok. +terminate(_Reason, #state{socket = undefined}) -> + ok; +terminate(_Reason, #state{socket = Socket}) -> + gen_tcp:close(Socket), + ok. + +-spec code_change(term(), #state{}, term()) -> {ok, #state{}}. +code_change(_OldVsn, State = #state{}, _Extra) -> + {ok, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec send_response(gen_tcp:socket(), {ok, iodata()} | {error, iodata()}) -> ok. +send_response(Socket, {ok, Response}) -> + Reply = json:encode(#{<<"result">> => Response}), + gen_tcp:send(Socket, Reply); +send_response(Socket, {error, Response}) -> + Reply = json:encode(#{<<"error">> => #{<<"message">> => Response}}), + gen_tcp:send(Socket, Reply). + +-spec dispatch(binary()) -> {ok, iodata()} | {error, iodata()}. +dispatch(Data) when is_binary(Data) -> + Request = json:decode(Data), + handle_request(Request). + +-spec handle_request(binary()) -> {ok, iodata()} | {error, iodata()}. +handle_request(#{<<"method">> := <<"ping">>}) -> + {ok, <<"pong">>}; + +handle_request(#{<<"method">> := <<"add_client">>, <<"params">> := #{<<"uuid">> := UUID, <<"token">> := Token}}) -> + case efka_client_store:register(UUID, Token) of + ok -> + {ok, <<"OK">>}; + {error, Reason} -> + ReadableReason = readable_binary(Reason), + {error, <<"add failed: ", ReadableReason/binary>>} + end; +handle_request(Command) -> + logger:warning("[ctrl_channel] unsupported command: ~p", [Command]), + {error, <<"error unsupported_command\n">>}. + +-spec readable_binary(Term :: any()) -> binary(). +readable_binary(Term) -> + iolist_to_binary(io_lib:format("~p", [Term])). diff --git a/apps/iot/src/ctrl/ctrl_server.erl b/apps/iot/src/ctrl/ctrl_server.erl new file mode 100644 index 0000000..272fb27 --- /dev/null +++ b/apps/iot/src/ctrl/ctrl_server.erl @@ -0,0 +1,111 @@ +%%%------------------------------------------------------------------- +%% @doc Unix domain socket control service. +%% @end +%%%------------------------------------------------------------------- + +-module(ctrl_server). +-behaviour(gen_server). + +-export([start_link/0]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). +-define(DEFAULT_SOCKET_PATH, "/var/lib/iot/ctl.sock"). +-define(DEFAULT_BACKLOG, 128). + +-record(state, { + listen_socket :: gen_tcp:socket(), + socket_path :: file:filename_all(), + acceptor :: pid() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +-spec start_link() -> {ok, pid()} | ignore | {error, term()}. +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + +%%%=================================================================== +%%% gen_server callbacks +%%%=================================================================== + +-spec init([]) -> {ok, #state{}} | {stop, term()}. +init([]) -> + ok = iot_log:set_metadata(), + Props = application:get_env(iot, ctrl_server, []), + SocketPath = proplists:get_value(socket_path, Props, ?DEFAULT_SOCKET_PATH), + Backlog = proplists:get_value(backlog, Props, ?DEFAULT_BACKLOG), + + ok = ensure_socket_dir(SocketPath), + ok = delete_stale_socket(SocketPath), + + ListenOpts = [ + binary, + {packet, line}, + {active, false}, + {backlog, Backlog}, + {ifaddr, {local, SocketPath}} + ], + case gen_tcp:listen(0, ListenOpts) of + {ok, ListenSocket} -> + {ok, Acceptor} = ctrl_acceptor:start_link(ListenSocket), + logger:debug("[ctrl_server] start at socket: ~ts", [SocketPath]), + {ok, #state{listen_socket = ListenSocket, socket_path = SocketPath, acceptor = Acceptor}}; + {error, Reason} -> + logger:error("[ctrl_server] failed to listen on socket ~ts: ~p", [SocketPath, Reason]), + {stop, Reason} + end. + +-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}. +handle_call(_Request, _From, State = #state{}) -> + {reply, ok, State}. + +-spec handle_cast(term(), #state{}) -> {noreply, #state{}}. +handle_cast(_Request, State = #state{}) -> + {noreply, State}. + +-spec handle_info(term(), #state{}) -> {noreply, #state{}}. +handle_info(Info, State = #state{}) -> + logger:warning("[ctrl_server] ignore unknown info: ~p", [Info]), + {noreply, State}. + +-spec terminate(term(), #state{}) -> ok. +terminate(_Reason, #state{listen_socket = ListenSocket, socket_path = SocketPath}) -> + gen_tcp:close(ListenSocket), + ok = delete_stale_socket(SocketPath), + ok. + +-spec code_change(term(), #state{}, term()) -> {ok, #state{}}. +code_change(_OldVsn, State = #state{}, _Extra) -> + {ok, State}. + +%%%=================================================================== +%%% Internal functions +%%%=================================================================== + +-spec ensure_socket_dir(file:filename_all()) -> ok | {error, term()}. +ensure_socket_dir(SocketPath) -> + case filelib:ensure_dir(SocketPath) of + ok -> + ok; + {error, Reason} -> + logger:error("[ctrl_server] failed to ensure socket dir for ~ts: ~p", + [SocketPath, Reason]), + {error, Reason} + end. + +-spec delete_stale_socket(file:filename_all()) -> ok. +delete_stale_socket(SocketPath) -> + case file:delete(SocketPath) of + ok -> + ok; + {error, enoent} -> + ok; + {error, Reason} -> + logger:warning("[ctrl_server] failed to delete stale socket ~ts: ~p", + [SocketPath, Reason]), + ok + end. diff --git a/apps/iot/src/iot_sup.erl b/apps/iot/src/iot_sup.erl index d12f3a4..610ea54 100644 --- a/apps/iot/src/iot_sup.erl +++ b/apps/iot/src/iot_sup.erl @@ -46,6 +46,15 @@ init([]) -> modules => ['udp_server'] }, + #{ + id => 'ctrl_server', + start => {'ctrl_server', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['ctrl_server'] + }, + #{ id => 'iot_host_sup', start => {'iot_host_sup', start_link, []}, diff --git a/config/sys.config.src b/config/sys.config.src index 0b3a0ec..68d4b1b 100644 --- a/config/sys.config.src +++ b/config/sys.config.src @@ -18,6 +18,11 @@ {port, 24000} ]}, + {ctrl_server, [ + {socket_path, "${IOT_CTRL_SOCKET_PATH:-/var/lib/iot/ctl.sock}"}, + {backlog, ${IOT_CTRL_SOCKET_BACKLOG:-128}} + ]}, + {api_url, "${IOT_API_URL}"} ]}, diff --git a/go_extend/.gitignore b/go_extend/.gitignore new file mode 100644 index 0000000..5849071 --- /dev/null +++ b/go_extend/.gitignore @@ -0,0 +1 @@ +/iot_ctrl diff --git a/go_extend/Makefile b/go_extend/Makefile new file mode 100644 index 0000000..bd34288 --- /dev/null +++ b/go_extend/Makefile @@ -0,0 +1,16 @@ +GO ?= go +GOCACHE ?= /private/tmp/go-build-cache +BINARY ?= iot_ctrl +GO_LDFLAGS ?= -linkmode=external + +.PHONY: build +build: + GOCACHE=$(GOCACHE) $(GO) build -ldflags="$(GO_LDFLAGS)" -o $(BINARY) ./cmd/iot_ctrl + +.PHONY: test +test: + GOCACHE=$(GOCACHE) $(GO) test -ldflags="$(GO_LDFLAGS)" ./cmd/iot_ctrl + +.PHONY: clean +clean: + rm -f $(BINARY) diff --git a/go_extend/cmd/iot_ctrl/iot_ctrl b/go_extend/cmd/iot_ctrl/iot_ctrl new file mode 100755 index 0000000..80572e0 Binary files /dev/null and b/go_extend/cmd/iot_ctrl/iot_ctrl differ diff --git a/go_extend/cmd/iot_ctrl/main.go b/go_extend/cmd/iot_ctrl/main.go new file mode 100644 index 0000000..bbaface --- /dev/null +++ b/go_extend/cmd/iot_ctrl/main.go @@ -0,0 +1,182 @@ +package main + +import ( + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "net" + "os" + "time" +) + +const ( + defaultSocketPath = "/var/lib/iot/ctl.sock" + maxPacketSize = 65535 + dialTimeout = 5 * time.Second + requestTimeout = 10 * time.Second +) + +type request struct { + Method string `json:"method"` + Params map[string]string `json:"params,omitempty"` +} + +type response struct { + Result json.RawMessage `json:"result,omitempty"` + Error *responseError `json:"error,omitempty"` +} + +type responseError struct { + Message string `json:"message"` +} + +func main() { + if err := run(os.Args[1:]); err != nil { + fmt.Fprintf(os.Stderr, "iot_ctrl: %v\n", err) + os.Exit(1) + } +} + +func run(args []string) error { + flags := flag.NewFlagSet("iot_ctrl", flag.ContinueOnError) + flags.SetOutput(io.Discard) + + socketPath := defaultSocketPath + if envSocketPath := os.Getenv("IOT_CTRL_SOCKET_PATH"); envSocketPath != "" { + socketPath = envSocketPath + } + flags.StringVar(&socketPath, "socket", socketPath, "control socket path") + + if err := flags.Parse(args); err != nil { + return usageError() + } + + rest := flags.Args() + if len(rest) == 0 { + return usageError() + } + + req, err := parseCommand(rest) + if err != nil { + return err + } + + result, err := call(socketPath, req) + if err != nil { + return err + } + if result != "" { + fmt.Println(result) + } + return nil +} + +func parseCommand(args []string) (request, error) { + switch args[0] { + case "ping": + if len(args) != 1 { + return request{}, usageError() + } + return request{Method: "ping"}, nil + case "add-client": + flags := flag.NewFlagSet("add-client", flag.ContinueOnError) + flags.SetOutput(io.Discard) + + var uuid string + var token string + flags.StringVar(&uuid, "uuid", "", "client uuid") + flags.StringVar(&token, "token", "", "client token") + if err := flags.Parse(args[1:]); err != nil { + return request{}, usageError() + } + if uuid == "" || token == "" || flags.NArg() != 0 { + return request{}, usageError() + } + return request{ + Method: "add_client", + Params: map[string]string{ + "uuid": uuid, + "token": token, + }, + }, nil + default: + return request{}, usageError() + } +} + +func call(socketPath string, req request) (string, error) { + conn, err := net.DialTimeout("unix", socketPath, dialTimeout) + if err != nil { + return "", err + } + defer conn.Close() + + if err := conn.SetDeadline(time.Now().Add(requestTimeout)); err != nil { + return "", err + } + + body, err := json.Marshal(req) + if err != nil { + return "", err + } + if err := writePacket(conn, body); err != nil { + return "", err + } + + reply, err := readPacket(conn) + if err != nil { + return "", err + } + + var resp response + if err := json.Unmarshal(reply, &resp); err != nil { + return "", err + } + if resp.Error != nil { + return "", errors.New(resp.Error.Message) + } + if len(resp.Result) == 0 { + return "", nil + } + + var text string + if err := json.Unmarshal(resp.Result, &text); err == nil { + return text, nil + } + return string(resp.Result), nil +} + +func writePacket(w io.Writer, body []byte) error { + if len(body) > maxPacketSize { + return fmt.Errorf("request exceeds packet=2 limit: %d bytes", len(body)) + } + + header := make([]byte, 2) + binary.BigEndian.PutUint16(header, uint16(len(body))) + if _, err := w.Write(header); err != nil { + return err + } + _, err := w.Write(body) + return err +} + +func readPacket(r io.Reader) ([]byte, error) { + header := make([]byte, 2) + if _, err := io.ReadFull(r, header); err != nil { + return nil, err + } + + size := binary.BigEndian.Uint16(header) + body := make([]byte, int(size)) + if _, err := io.ReadFull(r, body); err != nil { + return nil, err + } + return body, nil +} + +func usageError() error { + return errors.New("usage: iot_ctrl [--socket path] ping | iot_ctrl [--socket path] add-client --uuid --token ") +} diff --git a/go_extend/cmd/iot_ctrl/main_test.go b/go_extend/cmd/iot_ctrl/main_test.go new file mode 100644 index 0000000..702e231 --- /dev/null +++ b/go_extend/cmd/iot_ctrl/main_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "testing" +) + +func TestParsePingCommand(t *testing.T) { + req, err := parseCommand([]string{"ping"}) + if err != nil { + t.Fatalf("parseCommand returned error: %v", err) + } + if req.Method != "ping" { + t.Fatalf("method = %q, want ping", req.Method) + } + if req.Params != nil { + t.Fatalf("params = %#v, want nil", req.Params) + } +} + +func TestParseAddClientCommand(t *testing.T) { + req, err := parseCommand([]string{"add-client", "--uuid", "abcdefg", "--token", "token1234"}) + if err != nil { + t.Fatalf("parseCommand returned error: %v", err) + } + if req.Method != "add_client" { + t.Fatalf("method = %q, want add_client", req.Method) + } + if req.Params["uuid"] != "abcdefg" || req.Params["token"] != "token1234" { + t.Fatalf("params = %#v", req.Params) + } +} + +func TestPacket2RoundTrip(t *testing.T) { + req := request{Method: "ping"} + body, err := json.Marshal(req) + if err != nil { + t.Fatalf("json.Marshal returned error: %v", err) + } + + var buf bytes.Buffer + if err := writePacket(&buf, body); err != nil { + t.Fatalf("writePacket returned error: %v", err) + } + if got, want := binary.BigEndian.Uint16(buf.Bytes()[:2]), uint16(len(body)); got != want { + t.Fatalf("packet size = %d, want %d", got, want) + } + + got, err := readPacket(&buf) + if err != nil { + t.Fatalf("readPacket returned error: %v", err) + } + if !bytes.Equal(got, body) { + t.Fatalf("packet body = %q, want %q", got, body) + } +} diff --git a/go_extend/go.mod b/go_extend/go.mod new file mode 100644 index 0000000..8b787a2 --- /dev/null +++ b/go_extend/go.mod @@ -0,0 +1,3 @@ +module cloudkit/iot/go_extend + +go 1.21