add go command support

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.

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])).

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.

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, []},

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}"}
   ]},
 

go_extend/.gitignore

@@ -0,0 +1 @@
+/iot_ctrl

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)

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 <uuid> --token <token>")
+}

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)
+	}
+}

go_extend/go.mod

@@ -0,0 +1,3 @@
+module cloudkit/iot/go_extend
+
+go 1.21