[WIP] Output Embedded metrics for CloudWatch

aws/cloud-formation.template.yaml

@@ -306,6 +306,9 @@ Resources:
                             }
                           ]
                         }
+                      },
+                      "metrics_collected": {
+                        "emf": {}
                       }
                     }
                   }

config/sys.config

@@ -10,7 +10,13 @@
                     },
                 formatter =>
                     {logger_formatter, #{
-                        template => [time, " [", level, "] ", file, ":", line, " ", pid, " ", msg, "\n"]
+                        template => [
+                            {message_only,
+                                [],
+                                [time, " [", level, "] ", file, ":", line, " ", pid, " "]
+                            },
+                            msg, "\n"
+                        ]
                     }}
             }}
         ]}

src/Constants.sest

@@ -21,4 +21,7 @@ module Constants = struct
   val maximum_num_rooms_per_user() =
     10
 
+  val metrics_interval() =
+    1000 * 60 /* one minute */
+
 end

src/MetricsLogger.sest

@@ -0,0 +1,36 @@
+module MetricsLogger :> sig
+
+  type unix_timestamp_milliseconds = int
+
+  val put<$a> :
+    fun(
+      -timestamp unix_timestamp_milliseconds,
+      -num_rooms int,
+      -num_users int,
+    ) ->
+      [$a]unit
+
+end = struct
+  open Stdlib
+
+  type unix_timestamp_milliseconds = int
+
+  val put_impl<$a> : fun(list<char>) -> [$a]unit = external 1 ```
+    put_impl(Msg) ->
+        logger:log(info, Msg, [], #{ message_only => "yes", file => "", line => 0 }).
+  ```
+
+  val put(
+    -timestamp timestamp,
+    -num_rooms num_rooms,
+    -num_users num_users,
+  ) = act
+    let data =
+      format(
+        f'{\"_aws\": {\"Timestamp\": ~p, \"CloudWatchMetrics\": [{\"Namespace\": \"game-server-metrics\", \"Dimensions\": [[\"functionVersion\"]], \"Metrics\": [{\"Name\": \"num_rooms\": \"Unit\": \"Count\"}, {\"Name\": \"num_users\", \"Unit\": \"Count\"}]}]}, \"functionVersion\": \"$LATEST\", \"num_rooms\": ~p, \"num_users\": ~p}',
+        {timestamp, num_rooms, num_users},
+      )
+    in
+    put_impl(data)
+
+end

src/MetricsServer.sest

@@ -0,0 +1,137 @@
+import Logger
+import Constants
+import Socket
+import RoomResourceServer
+import UserResourceServer
+import MetricsLogger
+
+module MetricsServer :> sig
+  open Stdlib
+
+  type info :: o
+
+  type proc :: o
+
+  val start_link<$a> : fun() -> [$a]result<proc, GenServer.start_link_error>
+
+  val as_pid : fun(proc) -> pid<info>
+
+end = struct
+  open Stdlib
+
+  module Callback = struct
+
+    type init_arg = unit
+
+    type request =
+      | RequestDummy
+
+    type response =
+      | ResponseDummy
+
+    type cast_message =
+      | CastDummy
+
+    type info =
+      | InfoDummy
+
+    type state = {
+      socket : option<Socket.t>,
+    }
+
+    type global = unit
+
+    val init({}) = act
+      do res <- Socket.connect(-address {127, 0, 0, 1}, -port 25888) in
+      let socket_opt =
+        case res of
+        | Ok(socket) ->
+            Some(socket)
+        | Error(err) ->
+            let _ = assert Logger.warning(f'failed to connect (reason: ~p)', {err}) in
+            None
+        end
+      in
+      let state = { socket = socket_opt } in
+      let _ = assert Logger.info(f'start (state: ~p)', {state}) in
+      GenServer.init_ok(state, ?timeout Constants.metrics_interval())
+
+    val handle_call(RequestDummy, from, state) = act
+      let _ = assert Logger.warning(f'unexpected call (from: ~p, state: ~p)', {from, state}) in
+      GenServer.reply(ResponseDummy, state)
+
+    val handle_cast(CastDummy, state) = act
+      let _ = assert Logger.warning(f'unexpected cast (state: ~p)', {state}) in
+      GenServer.no_reply(state)
+
+    val handle_down(mref, pid, reason, state) = act
+      let tuple = {mref, pid, reason} in
+      let _ = assert Logger.warning(f'unexpected down (tuple: ~p, state: ~p)', {tuple, state}) in
+      GenServer.no_reply(state)
+
+    val now_milliseconds<$a> : fun() -> [$a]int = external 0 ```
+      now_milliseconds() ->
+          os:system_time(milli_seconds).
+    ```
+
+    val handle_timeout(state : state) = act
+      do res_rooms <- RoomResourceServer.get_number_of_rooms() in
+      do res_users <- UserResourceServer.get_number_of_users() in
+      do {} <-
+        case {res_rooms, res_users} of
+        | {Ok(num_rooms), Ok(num_users)} ->
+            let _ = assert Logger.debug(f'num_rooms: ~p, num_users: ~p', {num_rooms, num_users}) in
+            do timestamp <- now_milliseconds() in
+            do {} <-
+              MetricsLogger.put(
+                -timestamp timestamp,
+                -num_rooms num_rooms,
+                -num_users num_users,
+              )
+            in
+/*
+            case state.socket of
+            | Some(socket) ->
+                do res <- Socket.send(socket, Binary.from_list(data)) in
+                case res of
+                | Ok({}) ->
+                    return({})
+                | Error(err) ->
+                    let _ = assert Logger.warning(f'failed to send metrics (reason: ~p)', {err}) in
+                    return({})
+                end
+            | None ->
+                return({})
+            end
+*/
+            return({})
+        | _ ->
+            let _ = assert Logger.warning(f'res_rooms: ~p, res_users: ~p', {res_rooms, res_users}) in
+            return({})
+        end
+      in
+      GenServer.no_reply(state, ?timeout Constants.metrics_interval())
+
+    val handle_info(info, state : state) = act
+      let _ = assert Logger.warning(f'unexpected info (info: ~p, state: ~p)', {info, state}) in
+      GenServer.no_reply(state)
+
+    val terminate(reason, state) = act
+      let _ = assert Logger.warning(f'terminate (reason: ~p, state: ~p)', {reason, state}) in
+      return({})
+
+  end
+
+  module Impl = GenServer.Make(Callback)
+
+  type info = Callback.info
+
+  type proc = Impl.proc
+
+  val start_link() = act
+    Impl.start_link_name({}, -name Global({}))
+
+  val as_pid(proc) =
+    Impl.as_pid(proc)
+
+end

src/RoomResourceServer.sest

@@ -14,6 +14,7 @@ module RoomResourceServer :> sig
   val as_pid : fun(proc) -> pid<info>
   val add<$a, $b> :
     fun(-created_by user_id, -room_id room_id, -room_name binary) -> [$a]result<RoomServer.proc, RawValue.t>
+  val get_number_of_rooms<$a> : fun() -> [$a]result<int, RawValue.t>
 end = struct
   open Stdlib
   open Models
@@ -24,9 +25,11 @@ end = struct
 
     type request =
       | AddRoom(user_id, room_id, binary)
+      | GetNumRooms
 
     type response =
       | RoomAdded(result<RoomServer.proc, RawValue.t>)
+      | NumRoomsGot(int)
 
     type cast_message =
       | CastDummy
@@ -72,6 +75,9 @@ end = struct
             | Error(err) ->
                 GenServer.reply(RoomAdded(Error(err)), state)
             end
+
+      | GetNumRooms ->
+          GenServer.reply(NumRoomsGot(state.num_rooms), state)
       end
 
     val handle_cast(msg, state) = act
@@ -125,4 +131,17 @@ end = struct
         return(Error(RawValue.forget("no room resource server")))
     end
 
+  val get_number_of_rooms() = act
+    do proc_res <- Impl.where_is_global({}) in
+    case proc_res of
+    | Some(proc) ->
+        do call_res <- Impl.call(proc, Callback.GetNumRooms) in
+        case call_res of
+        | Ok(Callback.NumRoomsGot(n)) -> return(Ok(n))
+        | Error(err)                  -> return(Error(err))
+        end
+    | None ->
+        return(Error(RawValue.forget("no room resource server")))
+    end
+
 end

src/Socket.sest

@@ -0,0 +1,30 @@
+module Socket :> sig
+  open Stdlib
+
+  type t :: o
+
+  val connect<$a> : fun(-address {int, int, int, int}, -port int) -> [$a]result<t, RawValue.t>
+
+  val send<$a> : fun(t, binary) -> [$a]result<unit, RawValue.t>
+
+end = struct
+  open Stdlib
+
+  type t =
+    | SocketDummy
+
+  val connect<$a> : fun(-address {int, int, int, int}, -port int) -> [$a]result<t, RawValue.t> = external 2 ```
+    connect(Address, Port) ->
+        gen_tcp:connect(Address, Port, []).
+  ```
+
+  val send<$a> : fun(t, binary) -> [$a]result<unit, RawValue.t> = external 2 ```
+    send(Socket, Packet) ->
+        Result = gen_tcp:send(Socket, Packet),
+        case Result of
+            ok         -> {ok, ok};
+            {error, _} -> Result
+        end.
+  ```
+
+end

src/Sup.sest

@@ -3,6 +3,7 @@ import UserServerSup
 import PlazaServer
 import RoomResourceServer
 import UserResourceServer
+import MetricsServer
 
 module Sup = struct
   open Stdlib
@@ -46,6 +47,12 @@ module Sup = struct
         return(Result.map(UserResourceServer.as_pid, res))
       end)
 
+    val start_metrics_server() = act
+      SS.make_child_proc(fun() -> act
+        do res <- MetricsServer.start_link() in
+        return(Result.map(MetricsServer.as_pid, res))
+      end)
+
     val init(_) = act
       let sup_flags = SS.make_sup_flags(?strategy SS.OneForAll) in
       let child_specs =
@@ -75,6 +82,11 @@ module Sup = struct
             -start (freeze start_user_resource_server()),
             ?type  Supervisor.Worker
           ),
+          SS.make_child_spec(
+            -id    "metrics_server",
+            -start (freeze start_metrics_server()),
+            ?type  Supervisor.Worker
+          ),
         ]
       in
       SS.init_ok(sup_flags, child_specs)

src/UserResourceServer.sest

@@ -13,6 +13,7 @@ module UserResourceServer :> sig
   val start_link<$a> : fun() -> [$a]result<proc, GenServer.start_link_error>
   val as_pid : fun(proc) -> pid<info>
   val add<$a, $b> : fun(-user_id user_id, -user_name binary) -> [$a]result<UserServer.proc, RawValue.t>
+  val get_number_of_users<$a> : fun() -> [$a]result<int, RawValue.t>
 end = struct
   open Stdlib
   open Models
@@ -23,9 +24,11 @@ end = struct
 
     type request =
       | AddUser(user_id, binary)
+      | GetNumUsers
 
     type response =
       | UserAdded(result<UserServer.proc, RawValue.t>)
+      | NumUsersGot(int)
 
     type cast_message =
       | CastDummy
@@ -63,6 +66,9 @@ end = struct
             | Error(err) ->
                 GenServer.reply(UserAdded(Error(err)), state)
             end
+
+      | GetNumUsers ->
+          GenServer.reply(NumUsersGot(state.num_users), state)
       end
 
     val handle_cast(msg, state) = act
@@ -116,4 +122,17 @@ end = struct
         return(Error(RawValue.forget("no user resource server")))
     end
 
+  val get_number_of_users() = act
+    do proc_res <- Impl.where_is_global({}) in
+    case proc_res of
+    | Some(proc) ->
+        do call_res <- Impl.call(proc, Callback.GetNumUsers) in
+        case call_res of
+        | Ok(Callback.NumUsersGot(n)) -> return(Ok(n))
+        | Error(err)                  -> return(Error(err))
+        end
+    | None ->
+        return(Error(RawValue.forget("no user resource server")))
+    end
+
 end