HostNode statistics are missing timestamps

agent/lib/kontena/workers/node_info_worker.rb

@@ -135,7 +135,8 @@ def publish_node_stats
             used: disk.used_bytes,
             total: disk.total_bytes
           }
-        ]
+        ],
+        time: Time.now.utc.to_s
       }
       rpc_client.async.notification('/nodes/stats', [data])
       send_statsd_metrics(data)

agent/lib/kontena/workers/stats_worker.rb

@@ -93,6 +93,7 @@ def send_container_stats(container)
 
       current_stat = container.dig(:stats, -1)
       # Need to default to something usable in calculations
+
       cpu_usages = current_stat.dig(:cpu, :usage, :per_cpu_usage)
       num_cores = cpu_usages ? cpu_usages.count : 1
       raw_cpu_usage = current_stat.dig(:cpu, :usage, :total) - prev_stat.dig(:cpu, :usage, :total)
@@ -111,7 +112,8 @@ def send_container_stats(container)
         },
         filesystem: current_stat[:filesystem],
         diskio: current_stat[:diskio],
-        network: current_stat[:network]
+        network: current_stat[:network],
+        time: Time.now.utc.to_s
       }
       rpc_client.async.notification('/containers/stat', [data])
       send_statsd_metrics(name, data)

agent/spec/lib/kontena/workers/node_info_worker_spec.rb

@@ -214,4 +214,11 @@
       expect(subject.private_ip).to eq('192.168.2.10')
     end
   end
+
+  describe '#publish_node_stats' do
+    it 'sends stats via rpc with timestamps' do
+      expect(rpc_client).to receive(:notification).once.with('/nodes/stats', [hash_including(time: String)])
+      subject.publish_node_stats
+    end
+  end
 end

agent/spec/lib/kontena/workers/stats_workers_spec.rb

@@ -199,4 +199,33 @@
       subject.send_container_stats(event)
     end
   end
+
+  describe '#send_container_stats' do
+    it 'sends stats via rpc' do
+      expect(rpc_client).to receive(:notification).once.with('/containers/stat', [hash_including(time: String)])
+
+      container = {
+        aliases: [],
+        stats: [
+          {
+            timestamp: "2017-03-01 00:00:00",
+            cpu: {
+              usage: {
+                total: 1
+              }
+            }
+          },
+          {
+            timestamp: "2017-03-01 00:00:01",
+            cpu: {
+              usage: {
+                total: 1
+              }
+            }
+          }
+        ]
+      }
+      subject.send_container_stats(container)
+    end
+  end
 end

server/app/models/container_stat.rb

@@ -1,6 +1,6 @@
 class ContainerStat
   include Mongoid::Document
-  include Mongoid::Timestamps
+  include Mongoid::Timestamps::Created
 
   field :spec, type: Hash
   field :cpu, type: Hash

server/app/models/host_node_stat.rb

@@ -1,6 +1,6 @@
 class HostNodeStat
   include Mongoid::Document
-  include Mongoid::Timestamps
+  include Mongoid::Timestamps::Created
 
   field :load, type: Hash
   field :memory, type: Hash

server/app/services/rpc/container_handler.rb

@@ -7,11 +7,16 @@ class ContainerHandler
     include FixnumHelper
     include Logging
 
+    attr_accessor :logs_buffer_size
+    attr_accessor :stats_buffer_size
+
     def initialize(grid)
       @grid = grid
       @logs = []
       @stats = []
       @cached_container = nil
+      @logs_buffer_size = 5
+      @stats_buffer_size = 5
       @db_session = ContainerLog.collection.session.with(
         write: {
           w: 0, fsync: false, j: false
@@ -44,7 +49,7 @@ def log(data)
           type: data['type'],
           data: data['data']
         }
-        if @logs.size >= 5
+        if @logs.size >= @logs_buffer_size
           flush_logs
         end
       end
@@ -76,6 +81,7 @@ def stat(data)
       container = cached_container(data['id'])
       if container
         data = fixnums_to_float(data)
+        time = data['time'] ? Time.parse(data['time']) : Time.now.utc
         @stats << {
           grid_id: @grid.id,
           grid_service_id: container['grid_service_id'],
@@ -85,15 +91,20 @@ def stat(data)
           memory: data['memory'],
           filesystem: data['filesystem'],
           diskio: data['diskio'],
-          network: data['network']
+          network: data['network'],
+          created_at: time
         }
-        if @stats.size >= 5
-          @db_session[:container_stats].insert(@stats.dup)
-          @stats.clear
+        if @stats.size >= @stats_buffer_size
+          flush_stats
         end
       end
     end
 
+    def flush_stats
+      @db_session[:container_stats].insert(@stats.dup)
+      @stats.clear
+    end
+
     # @param [Hash] data
     def event(data)
       container = cached_container(data['id'])

server/app/services/rpc/node_handler.rb

@@ -7,6 +7,11 @@ class NodeHandler
 
     def initialize(grid)
       @grid = grid
+      @db_session = HostNode.collection.session.with(
+        write: {
+          w: 0, fsync: false, j: false
+        }
+      )
     end
 
     def get(id)
@@ -34,12 +39,18 @@ def stats(data)
       node = @grid.host_nodes.find_by(node_id: data['id'])
       return unless node
       data = fixnums_to_float(data)
-      node.host_node_stats.create(
+      time = data['time'] ? Time.parse(data['time']) : Time.now.utc
+
+      stat = {
         grid_id: @grid.id,
+        host_node_id: node.id,
         memory: data['memory'],
         load: data['load'],
-        filesystem: data['filesystem']
-      )
+        filesystem: data['filesystem'],
+        usage: data['usage'],
+        created_at: time
+      }
+      @db_session[:host_node_stats].insert(stat)
     end
   end
 end

server/spec/services/rpc/container_handler_spec.rb

@@ -107,23 +107,99 @@
       #puts bm
       total_time = Time.now.to_f - start_time
       expect(grid.container_logs.count).to eq(1_000)
-      expect(total_time < 0.5).to be_truthy
+      expect(total_time).to be < 0.5
     end
   end
 
-  describe '#on_container_health' do
-    it 'updates health status' do
+  describe '#stats_buffer_size' do
+    it 'has a default buffer size greater than 1' do
+      expect(subject.stats_buffer_size).to be > 1
+    end
+  end
+
+  describe '#logs_buffer_size' do
+    it 'has a default buffer size greater than 1' do
+      expect(subject.logs_buffer_size).to be > 1
+    end
+  end
+
+  describe '#on_stat' do
+    it 'saves container_stat items' do
+      subject.stats_buffer_size = 1
       container_id = SecureRandom.hex(16)
       container = grid.containers.new(name: 'foo-1', grid_service: grid_service)
       container.update_attribute(:container_id, container_id)
 
       expect {
-        subject.health({
+        subject.stat({
+          'id' => container_id,
+          'spec' => {},
+          'cpu' => {},
+          'memory' => {},
+          'filesystems' => [],
+          'diskio' => {},
+          'network' => {}
+        })
+      }.to change{ container.container_stats.count }.by 1
+    end
+
+    it 'buffers and saves container_stat items' do
+      buffer_size = subject.stats_buffer_size
+      container_id = SecureRandom.hex(16)
+      container = grid.containers.new(name: 'foo-1', grid_service: grid_service)
+      container.update_attribute(:container_id, container_id)
+
+      expect {
+        (buffer_size + 1).times {
+          subject.stat({
             'id' => container_id,
-            'status' => 'healthy'
-          }
-        )
-      }.to change {container.reload.health_status}.to eq('healthy')
+            'spec' => {},
+            'cpu' => {},
+            'memory' => {},
+            'filesystems' => [],
+            'diskio' => {},
+            'network' => {}
+          })
+        }
+      }.to change{ container.container_stats.count }.by buffer_size
+    end
+
+    it 'creates timestamps' do
+      subject.stats_buffer_size = 1
+      container_id = SecureRandom.hex(16)
+      container = grid.containers.new(name: 'foo-1', grid_service: grid_service)
+      container.update_attribute(:container_id, container_id)
+
+      subject.stat({
+          'id' => container_id,
+          'spec' => {},
+          'cpu' => {},
+          'memory' => {},
+          'filesystems' => [],
+          'diskio' => {},
+          'network' => {}
+      })
+      expect(container.container_stats[0].created_at).to be_a(Time)
+    end
+
+    it 'sets timestamps passed in' do
+      subject.stats_buffer_size = 1
+      container_id = SecureRandom.hex(16)
+      container = grid.containers.new(name: 'foo-1', grid_service: grid_service)
+      container.update_attribute(:container_id, container_id)
+      time = '2017-02-28 00:00:00 -0500'
+
+      subject.stat({
+          'id' => container_id,
+          'spec' => {},
+          'cpu' => {},
+          'memory' => {},
+          'filesystems' => [],
+          'diskio' => {},
+          'network' => {},
+          'time' => time
+      })
+      expect(container.container_stats[0].created_at).to eq Time.parse(time)
     end
   end
 end

server/spec/services/rpc/node_handler_spec.rb

@@ -20,5 +20,40 @@
         })
       }.to change{ node.host_node_stats.count }.by(1)
     end
+
+    it 'creates timestamps' do
+      node
+
+      subject.stats({
+        'node_id' => node.node_id,
+        'load' => {'1m' => 0.1, '5m' => 0.2, '15m' => 0.1},
+        'memory' => {},
+        'filesystems' => [],
+        'usage' => {
+          'container_seconds' => 60*100
+        }
+      })
+
+      expect(node.host_node_stats[0].created_at).to be_a(Time)
+    end
+
+    it 'sets timestamps passed in' do
+      node
+
+      time = '2017-02-28 00:00:00 -0500'
+
+      subject.stats({
+        'node_id' => node.node_id,
+        'load' => {'1m' => 0.1, '5m' => 0.2, '15m' => 0.1},
+        'memory' => {},
+        'filesystems' => [],
+        'usage' => {
+          'container_seconds' => 60*100
+        },
+        'time' => time
+      })
+
+      expect(node.host_node_stats[0].created_at).to eq Time.parse(time)
+    end
   end
 end