Added stats methods

Author: andrewosh

index.js

@@ -37,7 +37,9 @@ class HyperdriveDaemon extends EventEmitter {
 
     const corestoreOpts = {
       storage: path => raf(`${storage}/cores/${path}`),
-      sparse: true
+      sparse: true,
+      // Collect networking statistics.
+      stats: true
     }
     this.corestore = corestore(corestoreOpts.storage, corestoreOpts)
     // The root corestore should be bootstrapped with an empty default feed.

lib/drives/index.js

@@ -13,7 +13,8 @@ const {
   fromStat,
   toStat,
   toMount,
-  fromMount
+  fromMount,
+  toDriveStats
 } = require('hyperdrive-daemon-client/lib/common')
 const { rpc } = require('hyperdrive-daemon-client')
 
@@ -124,6 +125,41 @@ class DriveManager extends EventEmitter {
     this._sessions.delete(id)
   }
 
+  async getAllStats () {
+    const allStats = []
+    for (const [keyString, drive] of this._drives) {
+      const driveStats = await this.getDriveStats(drive)
+      allStats.push(driveStats)
+    }
+    return allStats
+  }
+
+  async getDriveStats (drive) {
+    const mounts = await drive.getAllMounts({ memory: true })
+    const stats = []
+
+    for (const { path, metadata, content} of mounts) {
+      stats.push({
+        path,
+        metadata: getCoreStats(metadata),
+        content: getCoreStats(content)
+      })
+    }
+
+    return stats
+
+    function getCoreStats (core) {
+      const stats = core.stats
+      return {
+        key: core.key,
+        peers: stats.peers.length,
+        uploadedBytes: stats.totals.uploadedBytes,
+        downloadedBytes: stats.totals.downloadedBytes
+        // TODO: Store comulative totals across restarts
+      }
+    }
+  }
+
   async get (key, opts) {
     key = (key instanceof Buffer) ? datEncoding.decode(key) : key
     var keyString = this._generateKeyString(key, opts)
@@ -224,6 +260,7 @@ function createDriveHandlers (driveManager) {
       const { drive, session } = await driveManager.createSession(driveOpts.key, driveOpts)
       driveOpts.key = drive.key
       driveOpts.version = drive.version
+      driveOpts.writable = drive.writable
 
       const rsp = new rpc.drive.messages.GetDriveResponse()
       rsp.setId(session)
@@ -232,6 +269,16 @@ function createDriveHandlers (driveManager) {
       return rsp
     },
 
+    allStats: async (call) => {
+      var stats = await driveManager.getAllStats()
+      stats = stats.map(driveStats => toDriveStats(driveStats))
+
+      const rsp = new rpc.drive.messages
+      rsp.setStats(stats)
+
+      return rsp
+    },
+
     publish: async (call) => {
       const id = call.request.getId()
 
@@ -256,6 +303,19 @@ function createDriveHandlers (driveManager) {
       return rsp
     },
 
+    stats: async (call) => {
+      const id = call.request.getId()
+
+      if (!id) throw new Error('A stats request must specify a session ID.')
+      const drive = driveManager.driveForSession(id)
+
+      const stats = await driveManager.getDriveStats(drive)
+
+      const rsp = new rpc.drive.messages.DriveStatsResponse()
+      rsp.setStats(toDriveStats(stats))
+      return rsp
+    },
+
     readFile: async (call) => {
       const id = call.request.getId()
       const path = call.request.getPath()
@@ -337,6 +397,61 @@ function createDriveHandlers (driveManager) {
       })
     },
 
+    mkdir: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+      const mode = call.request.getMode()
+
+      if (!id) throw new Error('A mkdir request must specify a session ID.')
+      if (!path) throw new Error('A mkdir request must specify a directory path.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.mkdir(path, mode, err => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.MkdirResponse()
+          return resolve(rsp)
+        })
+      })
+    },
+
+    rmdir: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+
+      if (!id) throw new Error('A rmdir request must specify a session ID.')
+      if (!path) throw new Error('A rmdir request must specify a directory path.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.rmdir(path, err => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.RmdirResponse()
+          return resolve(rsp)
+        })
+      })
+    },
+
+    unlink: async (call) => {
+      const id = call.request.getId()
+      const path = call.request.getPath()
+
+      if (!id) throw new Error('An unlink request must specify a session ID.')
+      if (!path) throw new Error('An unlink request must specify a path.')
+      const drive = driveManager.driveForSession(id)
+
+      return new Promise((resolve, reject) => {
+        drive.unlink(path, err => {
+          if (err) return reject(err)
+
+          const rsp = new rpc.drive.messages.UnlinkResponse()
+          return resolve(rsp)
+        })
+      })
+    },
+
     mount: async (call) => {
       const id = call.request.getId()
       const path = call.request.getPath()
@@ -357,15 +472,26 @@ function createDriveHandlers (driveManager) {
     },
 
     watch: async (call) => {
+      const id = call.request.getId()
+      var path = call.request.getPath()
 
-    },
-
-    listen: async (call) => {
-
-    },
-
-    unwatch: async (call) => {
+      if (!id) throw new Error('A watch request must specify a session ID.')
+      if (!path) path = '/'
+      const drive = driveManager.driveForSession(id)
 
+      const watcher = drive.watch(path, () => {
+        const rsp = new rpc.drive.messages.WatchResponse()
+        call.write(rsp)
+      })
+      call.on('close', onclose)
+      call.on('finish', onclose)
+      call.on('error', onclose)
+
+      function onclose (err) {
+        watcher.destroy()
+        log.debug({ id, path }, 'unregistering watcher')
+        if (err && !call.destroyed) call.destroy(err)
+      }
     },
 
     close: async (call) => {

package.json

@@ -25,6 +25,7 @@
   },
   "homepage": "https://github.com/andrewosh/hyperdrive-daemon#readme",
   "dependencies": {
+    "@grpc/grpc-js": "^0.5.1",
     "buffer-json-encoding": "^1.0.2",
     "chalk": "^2.4.2",
     "collect-stream": "^1.2.1",