feat: add stats to understand processing speeds

Author: stakach

src/source/publishing/influx_manager.cr

@@ -16,6 +16,10 @@ module PlaceOS::Source
 
     delegate start, stop, to: publisher
 
+    def stats : Hash(String, UInt64)
+      {"influx" => publisher.processed}
+    end
+
     def initialize(
       @influx_host : String = INFLUX_HOST || abort("INFLUX_HOST unset"),
       @influx_api_key : String = INFLUX_API_KEY || abort("INFLUX_API_KEY unset"),

src/source/publishing/mqtt_broker_manager.cr

@@ -26,6 +26,16 @@ module PlaceOS::Source
       end
     end
 
+    def stats : Hash(String, UInt64)
+      hash = {} of String => UInt64
+      read_publishers do |publishers|
+        publishers.values.each do |publisher|
+          hash["MQTT: #{publisher.broker.name}"] = publisher.processed
+        end
+      end
+      hash
+    end
+
     def process_resource(action : Resource::Action, resource : Model::Broker) : Resource::Result
       # Don't recreate the publisher if only "safe" attributes have changed
       case action

src/source/publishing/mqtt_publisher.cr

@@ -30,7 +30,7 @@ module PlaceOS::Source
       Event.new(value, timestamp).to_json
     end
 
-    private getter broker : PlaceOS::Model::Broker
+    getter broker : PlaceOS::Model::Broker
     private getter broker_lock : RWLock = RWLock.new
     protected getter client : ::MQTT::V3::Client
 

src/source/publishing/publisher.cr

@@ -11,26 +11,26 @@ module PlaceOS::Source
       timestamp : Time
     )
 
-    getter message_queue : Channel(Message) = Channel(Message).new
+    getter message_queue : Channel(Message) = Channel(Message).new(StatusEvents::BATCH_SIZE)
+    getter processed : UInt64 = 0_u64
 
     abstract def publish(message : Message)
 
     def start
-      consume_messages
+      spawn { consume_messages }
     end
 
     def stop
       message_queue.close
     end
 
     private def consume_messages
-      spawn do
-        while message = message_queue.receive?
-          begin
-            publish(message)
-          rescue error
-            Log.warn(exception: error) { "publishing message: #{message}" }
-          end
+      while message = message_queue.receive?
+        begin
+          publish(message)
+          @processed += 1_u64
+        rescue error
+          Log.warn(exception: error) { "publishing message: #{message}" }
         end
       end
     end

src/source/publishing/publisher_manager.cr

@@ -5,5 +5,7 @@ module PlaceOS::Source
     abstract def broadcast(message : Publisher::Message)
     abstract def start
     abstract def stop
+
+    abstract def stats : Hash(String, UInt64)
   end
 end

src/source/status_events.cr

@@ -14,7 +14,7 @@ module PlaceOS::Source
     STATUS_CHANNEL_PATTERN      = "status/#{Model::Module.table_name}-*"
     MAX_CONTAINER_SIZE          = 40_000
     BATCH_SIZE                  =    100
-    PROCESSING_INTERVAL         = 100.milliseconds
+    PROCESSING_INTERVAL         = 40.milliseconds
     CONTAINER_WARNING_THRESHOLD = MAX_CONTAINER_SIZE * 0.8
 
     private getter! redis : Redis
@@ -80,7 +80,9 @@ module PlaceOS::Source
 
         # process
         break if modules.empty?
-        yield modules
+        modules.each do |mod|
+          yield mod
+        end
         break if modules.size < batch_size
 
         last_created_at = modules.last.created_at
@@ -92,26 +94,23 @@ module PlaceOS::Source
       mods_mapped = 0_u64
       status_updated = 0_u64
       pattern = "initial_sync"
-      paginate_modules do |modules|
-        modules.each do |mod|
-          next unless mod
-          mods_mapped += 1_u64
-          module_id = mod.id.to_s
-          store = PlaceOS::Driver::RedisStorage.new(module_id)
-          store.each do |key, value|
-            status_updated += 1_u64
-            add_event({source: :db, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
-          end
+      paginate_modules do |mod|
+        mods_mapped += 1_u64
+        module_id = mod.id.to_s
+        store = PlaceOS::Driver::RedisStorage.new(module_id)
+        store.each do |key, value|
+          status_updated += 1_u64
+          add_event({source: :db, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
+        end
 
-          # Backpressure if event container is growing too fast
-          if event_container.size >= MAX_CONTAINER_SIZE / 2
-            until event_container.size < MAX_CONTAINER_SIZE / 4
-              sleep 10.milliseconds
-            end
+        # Backpressure if event container is growing too fast
+        if event_container.size > MAX_CONTAINER_SIZE // 2
+          until event_container.size < MAX_CONTAINER_SIZE // 4
+            sleep 10.milliseconds
           end
-        rescue error
-          Log.warn(exception: error) { "error syncing #{mod.try(&.id)}" }
         end
+      rescue error
+        Log.warn(exception: error) { "error syncing #{mod.try(&.id)}" }
       end
       Log.info { {
         message: "initial status sync complete",
@@ -131,26 +130,23 @@ module PlaceOS::Source
       status_updated = 0_u64
       pattern = "broker_resync"
 
-      paginate_modules do |modules|
-        modules.each do |mod|
-          next unless mod
-          mods_mapped += 1_u64
-          module_id = mod.id.to_s
-          store = PlaceOS::Driver::RedisStorage.new(module_id)
-          store.each do |key, value|
-            status_updated += 1_u64
-            add_event({source: :db, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
-          end
+      paginate_modules do |mod|
+        mods_mapped += 1_u64
+        module_id = mod.id.to_s
+        store = PlaceOS::Driver::RedisStorage.new(module_id)
+        store.each do |key, value|
+          status_updated += 1_u64
+          add_event({source: :db, mod_id: module_id, status: key}, {pattern: pattern, payload: value, timestamp: Time.utc})
+        end
 
-          # Backpressure if event container is growing too fast
-          if event_container.size >= MAX_CONTAINER_SIZE / 2
-            until event_container.size < MAX_CONTAINER_SIZE / 4
-              sleep 10.milliseconds
-            end
+        # Backpressure if event container is growing too fast
+        if event_container.size >= MAX_CONTAINER_SIZE // 2
+          until event_container.size < MAX_CONTAINER_SIZE // 4
+            sleep 10.milliseconds
           end
-        rescue error
-          Log.warn(exception: error) { "error resyncing #{mod.try(&.id)}" }
         end
+      rescue error
+        Log.warn(exception: error) { "error resyncing #{mod.try(&.id)}" }
       end
 
       Log.info { {
@@ -191,6 +187,18 @@ module PlaceOS::Source
           else
             process_batch(batch)
             Fiber.yield
+
+            # This outputs how many writes have occured for each publisher
+            # stats = String.build do |io|
+            #  io << "\n\n\nNEXT BATCH:\n"
+            #  publisher_managers.each do |manager|
+            #    manager.stats.each do |name, count|
+            #      io << "  * #{name} => #{count}"
+            #    end
+            #  end
+            #  io << "\n\n"
+            # end
+            # puts stats
           end
         rescue error
           Log.error(exception: error) { "error processing events" }