Add the keygrabber control keywords and a testable scheduler

docs/source/api/index.md

@@ -85,6 +85,12 @@ the docs build does not install the optional `influxdb` extra.
    :show-inheritance:
 ```
 
+```{eval-rst}
+.. automodule:: libby.keygrabber.scheduler
+   :members:
+   :show-inheritance:
+```
+
 ```{eval-rst}
 .. automodule:: libby.keygrabber.daemon
    :members:

docs/source/keygrabber.md

@@ -20,10 +20,9 @@ pip install libby[influxdb]
 keygrabber -c /etc/hispec/keygrabber.yaml
 ```
 
-It is an ordinary `LibbyDaemon`, so `SIGTERM` stops it cleanly and the
-`shutdown` keyword will too once the control surface lands. On the way out it
-gives its retry queue a bounded chance to drain, so a graceful stop does not
-lose the last tick.
+It is an ordinary `LibbyDaemon`, so `SIGTERM` stops it cleanly and so does
+writing its `shutdown` keyword. On the way out it gives its retry queue a
+bounded chance to drain, so a graceful stop does not lose the last tick.
 
 ## Config
 
@@ -99,6 +98,56 @@ and a tighter interval would be overrun by a single slow peer.
 A tick whose predecessor is still running is skipped rather than queued behind
 it, so a wedged peer cannot accumulate overlapping reads.
 
+## Control keywords
+
+The keygrabber is itself a peer, so its cadence and health are reachable with
+the ordinary `libby` verbs, and it can feed the same dashboards it fills.
+
+| Keyword | Type | Access | Meaning |
+|---|---|---|---|
+| `enabled` | bool | R/W | Collect on the configured cadences; false pauses without exiting |
+| `isconnected` | bool | R/W | Last sink write succeeded; write true to request a reconnect |
+| `pointswritten` | int | R | Samples the sink has stored since start |
+| `readerrors` | int | R | Keyword reads that failed |
+| `writeerrors` | int | R | Sink writes that failed |
+| `queuedepth` | int | R | Batches waiting in the retry queue |
+| `skippedticks` | int | R | Ticks skipped because the previous read was still in flight |
+| `droppedbatches` | int | R | Batches discarded because a queue was full |
+| `reload` | trigger | W | Re-read the config file and apply it |
+| `shutdown` | trigger | W | Gracefully stop the daemon |
+| `<collection>.enabled` | bool | R/W | Collect this one collection |
+| `<collection>.interval` | float | R/W | Cadence in seconds |
+| `<collection>.lastsample` | string | R | UTC time of the last successful tick |
+| `<collection>.lag` | float | R | Seconds the last tick ran past its due time |
+
+```bash
+libby show   hispec.keygrabber.%
+libby modify hispec.keygrabber.adc.interval=30
+libby modify hispec.keygrabber.enabled=false
+```
+
+Per-collection keywords contain a dot, and `%` matches within a single segment
+only, so `libby list hispec.keygrabber.%` will not show `adc.enabled`. Use
+`hispec.keygrabber.%.%` for those.
+
+`isconnected` reports whether the last write to the sink succeeded; it does not
+ping the database, because these getters are answered on the transport's
+receive thread and a blocking one would time out every read in flight. Writing
+`true` asks the writer thread to reconnect and returns immediately, so poll the
+keyword for the outcome. There is no manual disconnect.
+
+### reload
+
+`reload` re-reads the config file and applies it to the running collections,
+so keyword selections and cadences can change without a restart. A file that
+fails to parse leaves the running collections untouched and reports why, both
+to the caller and on `lasterror`.
+
+It will not add or remove collections, and says so rather than half-applying:
+libby has no way to withdraw a keyword, so a new collection's control keywords
+could not appear without a restart. Changing `sink` or `workers` also needs a
+restart.
+
 ## How it reads
 
 A tick is one `keys.read` request per peer, not one per keyword. This matters

libby/keygrabber/collection.py

@@ -22,7 +22,9 @@ class TickResult:
     read_errors: int
 
 
-class Collection:
+# Config plus the runtime state the control keywords expose; each attribute is
+# one reported value rather than hidden complexity
+class Collection:  # pylint: disable=too-many-instance-attributes
     """Tracks what one peer exposes and turns a read of it into samples.
 
     Resolution is refreshed periodically rather than once, so keywords added by
@@ -36,6 +38,13 @@ def __init__(
         clock: Callable[[], float] = time.monotonic,
     ) -> None:
         self.config = config
+        # Runtime state the control keywords read and write. Plain attributes
+        # rather than lock-guarded: each is a single value written by one
+        # thread and read by the transport's receive thread, and that thread
+        # must never block on a lock a tick might hold.
+        self.enabled = True
+        self.last_sample: Optional[datetime] = None
+        self.lag_s = 0.0
         self._clock = clock
         self._names: Tuple[str, ...] = ()
         self._bulk_read = False
@@ -62,6 +71,10 @@ def needs_resolve(self) -> bool:
             return True
         return self._clock() - self._resolved_at >= self.config.refresh_s
 
+    def invalidate(self) -> None:
+        """Force the next tick to resolve again, after a config change."""
+        self._resolved_at = None
+
     def resolve(self, client: Client) -> Tuple[str, ...]:
         """Ask the peer what it serves and select the configured keywords.