Commit local changes

docker-compose.yml

@@ -13,6 +13,7 @@ services:
       DB_USER: root
       DB_PASS: testing
       DB_DATABASE: agent
+      DEBUG_USER: ags131
     volumes:
     - './:/app'
     ports:
@@ -36,12 +37,12 @@ services:
   influxdb:
     image: influxdb
     volumes:
-      - './data/influxdb:/var/lib/influxdb'
+    - './data/influxdb:/var/lib/influxdb'
     restart: unless-stopped
   mysql:
     image: mysql:5.7
     volumes:
-      - './data/mysql:/var/lib/mysql'
+    - './data/mysql:/var/lib/mysql'
     environment:
       MYSQL_ROOT_PASSWORD: testing
     restart: unless-stopped

public/index.html

@@ -44,6 +44,13 @@
           <label>Interval (Should be 60 for Memory, 15 for segment)</label>
           <input type="number" ng-model="conf.methodOptions.interval" value="60">
         </div>
+        <div>
+          <label>Output</label>
+          <select ng-model="conf.output">
+            <option value="graphite">Graphite</option>
+            <option value="influxdb">InfluxDB</option>
+          </select>
+        </div>
       </div>
     </div>
     <script>
@@ -69,6 +76,7 @@
                 segment: '',
                 interval: 60
               },
+              output: 'graphite',
               create: true
             })
           }

src/drivers/output/graphite.js

@@ -48,7 +48,8 @@ export default async function handle ({ username, prefix = '' } = {}, { type, st
       if (!key || !value || !time) {
         continue
       }
-      if (Date.now() / 1000 < parseInt(time) - 1000) {
+      time = parseInt(time)
+      if (Date.now() / 1000 < time - 1000) {
         time /= 1000
       }
       value = parseFloat(value)
@@ -58,10 +59,12 @@ export default async function handle ({ username, prefix = '' } = {}, { type, st
   }
   if (type === 'application/json') {
     const ts = Math.round(Date.now() / 1000)
-    const data = flattenObj({}, `screeps.${username}`, stats)
+    let pre = `screeps.${username}.${prefix}`
+    if(pre.endsWith('.')) pre = pre.slice(0,-1)
+    const data = flattenObj({}, pre, stats)
     for(const key in data) {
       const value = Math.round(parseFloat(data[key]) * 1000) / 1000
-      const stat = `${prefix}${key} ${value} ${ts}`
+      const stat = `${key} ${value} ${ts}`
       out.push(stat)
     }
   }

src/drivers/output/index.js

@@ -1,9 +1,11 @@
-import Console from './console'
-import ScreepsPlus from './screepsplus'
-import Graphite from './graphite'
+import console from './console'
+import screepsplus from './screepsplus'
+import graphite from './graphite'
+import influxdb from './influxdb'
 
 export default {
-  Console,
-  ScreepsPlus,
-  Graphite
+  console,
+  screepsplus,
+  graphite,
+  influxdb
 }

src/drivers/output/influxdb.js

@@ -0,0 +1,82 @@
+import axios from 'axios'
+import { Agent } from 'http'
+
+const conn = new TCPConnection(process.env.GRAPHITE_HOST || 'graphite', process.env.GRAPHITE_PORT || '2003')
+
+let http = axios.create({
+  baseURL: process.env.INFLUXDB_URL || 'http://influxdb:8086',
+  httpAgent: new Agent({ keepAlive: true }),
+  auth: {
+    username: process.env.INFLUXDB_USERNAME,
+    password: process.env.INFLUXDB_PASSWORD
+  }
+})
+
+export default async function handle ({ username, prefix = '' } = {}, { type, stats }) {
+  if (!username) throw new Error('Username required')
+
+  async function query(strings, ...values) {
+    let sql = strings.shift()
+    for(const v of values) {
+      sql += values + strings.shift()
+    }
+    const { data } = await http.GET('/query', {
+      query: {
+        db: username,
+        q: sql
+      }
+    })
+    return data
+  }
+
+  const out = []
+  if (prefix && !prefix.endsWith('.')) prefix += '.'
+  if (type.match(/^text\/influxdb$/)) {
+    stats = stats.split("\n").filter(Boolean)
+    for (const stat of stats) {
+      let [, key, value, time] = stat.match(/^(\S+) (\S+)(?: (\d+))?$/) || []
+      if (!key || !value) {
+        continue
+      }
+      time = parseInt(time) || Date.now()
+      if (Date.now() / 1000 < time - 1000) {
+        time /= 1000
+      }
+      out.push(`${key} ${value} ${time}`)
+    }
+  }
+  if (type === 'application/json') {
+    const ts = Math.round(Date.now() / 1000)
+    let pre = prefix
+    if(pre.endsWith('.')) pre = pre.slice(0,-1)
+    const data = flattenObj({}, pre, stats)
+    for(const key in data) {
+      const value = Math.round(parseFloat(data[key]) * 1000) / 1000
+      const stat = `${key} value=${value} ${ts}`
+      out.push(stat)
+    }
+  }
+  console.log(`Writing ${out.length} stats for user ${username} with prefix '${prefix}'`)
+  await conn.write(out.join("\n") + "\n")
+
+  await query`CREATE DATABASE ${username} WITH DURATION 1w REPLICATION 1 NAME "1w_stats"`
+  await http.post('/write', {
+    query: {
+      db: username,
+      precision: 's',
+      rp: '1w_stats',
+    },
+    data: out.join("\n") + "\n"
+  }
+}
+
+function flattenObj (ret, path, obj) {
+  if (typeof obj == 'object') {
+    for (let k in obj) {
+      flattenObj(ret, path ? `${path}.${k}` : k, obj[k])
+    }
+  } else {
+    ret[path] = obj
+  }
+  return ret
+}

src/routes/index.js

@@ -28,6 +28,7 @@ async function cors(res) {
 }
 
 function checkAuth(req) {
+  if (process.env.DEBUG_USER) return process.env.DEBUG_USER
   return (req.headers['token-claim-sub'] || req.headers['token-claim-user'] || '').toLowerCase()
 }
 

src/work-manager.js

@@ -5,12 +5,12 @@ import { EventEmitter } from 'events'
 export class Worker extends EventEmitter {
   constructor() {
     super()
-    this.output = drivers.output.Graphite
   }
   async start (config) {
     let conf = config.methodConfig || {}
     const api = new ScreepsAPI(config.screepsAPIConfig)
     const driver = this.driver = new drivers.input[config.method](api, conf)
+    const output = this.output = drivers.output[config.output || 'graphite']
     driver.on('stats', async (stats) => {
       try {
         if(!stats) {