Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
27ed45999b | ||
|
|
9254f043a4 |
@@ -1,34 +1,42 @@
|
|||||||
# meta-core
|
# meta-core
|
||||||
|
|
||||||
Standalone Go service that owns Redis, the leader-info file, the HTTP+SSE
|
Standalone Go service that owns Redis, the HTTP+SSE metadata surface, the
|
||||||
metadata surface, and a WebDAV file server for the MetaMesh dev/prod stack.
|
User Data Layer store and signing keystore, and a WebDAV file server for the
|
||||||
|
MetaMesh dev/prod stack.
|
||||||
|
|
||||||
## Overview
|
## Overview
|
||||||
|
|
||||||
meta-core runs as its own container (`metacore-app`). Other MetaMesh services
|
meta-core runs as its own container (`metacore-app`). Other MetaMesh services
|
||||||
(meta-sort, meta-fuse, meta-stremio, meta-dup, meta-share) discover it by
|
(meta-sort, meta-fuse, meta-stremio, meta-dup, meta-search, meta-share,
|
||||||
reading `/meta-core/locks/kv-leader.info` and then call its HTTP API. Redis
|
meta-gateway, meta-watch, …) locate it over UDP multicast (beacon v2 —
|
||||||
is not exposed outside the container — all metadata I/O is mediated by the
|
its `metamesh.core` advertise carries the full `/urls` payload) and then call its
|
||||||
HTTP API.
|
HTTP API. `META_CORE_URL` on a client pins it instead. Redis is not published
|
||||||
|
outside the container — all metadata I/O is mediated by the HTTP/SSE API
|
||||||
|
(see [`docs/api-mediated-access.md`](docs/api-mediated-access.md)).
|
||||||
|
|
||||||
### Key capabilities
|
### Key capabilities
|
||||||
|
|
||||||
| Feature | Description |
|
| Feature | Description |
|
||||||
|---------|-------------|
|
|---------|-------------|
|
||||||
| **Leader gate** | Bash `flock` loop in `docker/leader-election.sh` only `exec`s supervisord on the winner. The Go binary therefore only ever runs as leader. |
|
| **Redis-volume mutex** | Bash `flock` loop in `docker/leader-election.sh` only `exec`s supervisord on the winner, so two containers sharing one `/meta-core` volume can never run two Redis writers. It is a mutex, not discovery — nothing reads the lock. |
|
||||||
| **Redis owner** | Supervisord starts Redis with AOF + RDB persistence inside the container; meta-core connects to it on localhost. |
|
| **Redis owner** | Supervisord starts Redis with AOF + RDB persistence inside the container; meta-core connects to it on localhost. |
|
||||||
| **HTTP API** | Typed REST surface (gorilla/mux) for metadata, KV browsing, snapshots, schema inference, mounts, watchers, neighbours, and file access by CID. |
|
| **HTTP API** | Typed REST surface (gorilla/mux) for metadata, KV browsing, snapshots, schema inference, mounts, watchers, neighbours, identity, UDL, and file access by CID. |
|
||||||
|
| **In-memory search index** | `POST /api/metadata/search` matches against an in-memory snapshot of every record, kept current by incremental refreshes (full reconcile every 30 min) — no Redis round-trip per record on the request path (`internal/api/search_index.go`). |
|
||||||
| **SSE event streams** | `/api/events/files` and `/api/events/meta` mirror the underlying Redis Streams so external services never speak Redis. |
|
| **SSE event streams** | `/api/events/files` and `/api/events/meta` mirror the underlying Redis Streams so external services never speak Redis. |
|
||||||
| **WebDAV server** | `/webdav/*` exposes the `/files` volume (read+write) for cross-service file access. nginx in front of meta-core handles caching. |
|
| **WebDAV server** | `/webdav/*` exposes the `/files` volume (read+write) for cross-service file access. nginx in front of meta-core handles caching. |
|
||||||
| **Service discovery** | meta-discovery v1: UDP multicast announce/listen on `239.255.77.1:9399`, neighbours served at `/neighbors`. The file-based registry under `/meta-core/services/*.json` it replaced is gone. |
|
| **Service discovery** | beacon v2: UDP multicast advertise/listen on `239.255.99.1:9099`, neighbours served at `/neighbors`. The file-based registry under `/meta-core/services/*.json` it replaced is gone. |
|
||||||
| **File watcher / scanner** | Recursive scan + MidHash256 computation; emits events on `file:events`. |
|
| **File watcher / scanner** | Recursive scan + MidHash256 computation; emits events on `file:events`. Watched roots are POSTed watcher configs (`/meta-core/watchers.json`), not env vars. |
|
||||||
| **Mount management** | rclone-only mount manager (SMB rendered into `:smb:`, plus pre-defined remotes). Read-only by construction. |
|
| **Mount management** | rclone-only mount manager (SMB rendered into `:smb:`, plus pre-defined remotes). Read-only by construction. |
|
||||||
| **UUID-rooted storage** | Roots are UUIDv7 (Crockford Base32, ULID layout); CIDs are reverse-index aliases. Schema-version sentinel refuses to boot against stale data. |
|
| **UUID-rooted storage** | Roots are UUIDv7 (Crockford Base32, ULID layout); CIDs are reverse-index aliases. Schema-version sentinel refuses to boot against stale data. |
|
||||||
|
| **Identity keystore** | Multi-account secp256k1 signing keys under `/meta-core/identity/accounts/`, created from the dashboard's Identity menu (never auto-generated). A private key is returned only by generate, or by reveal against a proof-of-possession signature (delete needs the same proof); `/api/identity/sign` refuses challenge-domain payloads. |
|
||||||
|
| **User Data Layer** | Version-gated store for per-user signed (and, for private-tier keys, encrypted) records — My List, Continue Watching, ratings — indexed by user and CID. meta-core never inspects the crypto. See [User Data Layer](../../docs/project-architecture/user-data-layer.md). |
|
||||||
|
| **Dashboard + editor** | nginx serves a React dashboard at `/` and the standalone metadata editor at `/editor/`. |
|
||||||
|
|
||||||
### Design characteristics
|
### Design characteristics
|
||||||
|
|
||||||
- **Single-leader** — flock on a shared volume; no external consensus.
|
- **Single writer** — flock on the `/meta-core` volume; no external consensus.
|
||||||
- **HTTP-only externally** — Redis is bound to the container, never published.
|
- **HTTP-only externally** — Redis is never published by the compose stacks;
|
||||||
|
debug with `docker exec metacore-app redis-cli …`.
|
||||||
- **Schema-gated boot** — `EnsureSchemaVersion` aborts startup when the
|
- **Schema-gated boot** — `EnsureSchemaVersion` aborts startup when the
|
||||||
on-disk Redis layout predates the current build (alpha clean-wipe policy;
|
on-disk Redis layout predates the current build (alpha clean-wipe policy;
|
||||||
see `docs/uuid-rooted-metadata.md`).
|
see `docs/uuid-rooted-metadata.md`).
|
||||||
@@ -41,34 +49,31 @@ HTTP API.
|
|||||||
┌──────────────────────────────────────────────────────────────────────────┐
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
│ metacore-app (standalone container) │
|
│ metacore-app (standalone container) │
|
||||||
│ │
|
│ │
|
||||||
|
│ leader-election.sh (flock /meta-core/locks/kv-leader.lock, │
|
||||||
|
│ then exec supervisord) │
|
||||||
│ supervisord │
|
│ supervisord │
|
||||||
│ ├─ leader-election.sh (flock /meta-core/locks/kv-leader.lock, │
|
|
||||||
│ │ writes kv-leader.info, exec supervisord) │
|
|
||||||
│ ├─ redis-server (AOF + RDB on /meta-core/db/redis) │
|
│ ├─ redis-server (AOF + RDB on /meta-core/db/redis) │
|
||||||
│ ├─ nginx (TLS termination, proxy_cache for /webdav) │
|
│ ├─ nginx :80 (dashboard, /editor, API proxy, │
|
||||||
│ ├─ rclone rcd (mount daemon, RC API) │
|
│ │ proxy_cache for /webdav, rclone GUI /rclone) │
|
||||||
|
│ ├─ rclone rcd (mount daemon, RC API on :5572) │
|
||||||
│ ├─ mount-watcher.sh (rclone mount lifecycle) │
|
│ ├─ mount-watcher.sh (rclone mount lifecycle) │
|
||||||
│ └─ meta-core (Go) ─► HTTP API + SSE + WebDAV on :9000 │
|
│ └─ meta-core (Go) ─► HTTP API + SSE + WebDAV on :9000 │
|
||||||
|
│ ─► UDP advertise/listen 239.255.99.1:9099 │
|
||||||
└──────────────────────────────────────────────────────────────────────────┘
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
│
|
│
|
||||||
┌──────────────────────┴──────────────────────┐
|
┌──────────────────────┴──────────────────────┐
|
||||||
│ /meta-core (shared volume) │
|
│ /meta-core (meta-core's own volume) │
|
||||||
│ locks/kv-leader.{lock,info} │
|
│ locks/kv-leader.lock │
|
||||||
│ db/redis/ │
|
│ db/redis/ │
|
||||||
│ services/*.json │
|
│ identity/accounts/*.json │
|
||||||
│ mounts/{mounts.json, errors/} │
|
│ mounts/{mounts.json, errors/} │
|
||||||
│ watchers.json │
|
│ watchers.json │
|
||||||
│ cache/ (nginx proxy_cache) │
|
│ cache/ (nginx proxy_cache) │
|
||||||
└─────────────────────────────────────────────┘
|
└─────────────────────────────────────────────┘
|
||||||
▲
|
|
||||||
┌──────────────────────────────┼────────────────────────────────┐
|
meta-sort / meta-fuse / meta-stremio / meta-search / meta-share / …
|
||||||
│ │ │
|
│ 1. hear the metamesh.core advertise on UDP multicast (carries /urls)
|
||||||
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
|
└─ 2. HTTP + SSE (+ WebDAV) to metacore-app — never Redis, never the volume
|
||||||
│ meta-sort │ │ meta-fuse │ │ meta-stremio│
|
|
||||||
│ │ │ │ │ │
|
|
||||||
│ reads │ HTTP+SSE │ reads │ HTTP+SSE │ reads │
|
|
||||||
│ leader.info├──────────────►│ leader.info├───────────────►│ leader.info│
|
|
||||||
└─────────────┘ └─────────────┘ └─────────────┘
|
|
||||||
```
|
```
|
||||||
|
|
||||||
## Quick start
|
## Quick start
|
||||||
@@ -82,8 +87,9 @@ make test # run Go unit tests
|
|||||||
```
|
```
|
||||||
|
|
||||||
`make docker` tags the image as `:1.0.0`, `:latest`, **and** `:local`. The
|
`make docker` tags the image as `:1.0.0`, `:latest`, **and** `:local`. The
|
||||||
`:local` tag is the one every sub-stack docker-compose.yml references
|
`:local` tag is the one the sub-stack docker-compose files reference
|
||||||
(meta-share 10-peer federation, meta-gateway dev stacks); without it the
|
(meta-share federation, meta-read and meta-listen dev stacks); the meta-gateway
|
||||||
|
dev stack (`docker-compose.gateway.yml`) uses `:latest`. Without these tags the
|
||||||
sub-stacks can't see a freshly-built image by name.
|
sub-stacks can't see a freshly-built image by name.
|
||||||
|
|
||||||
> **Source-change → recreate.** `make docker` rebuilds and re-tags the
|
> **Source-change → recreate.** `make docker` rebuilds and re-tags the
|
||||||
@@ -94,12 +100,14 @@ sub-stacks can't see a freshly-built image by name.
|
|||||||
>
|
>
|
||||||
> ```bash
|
> ```bash
|
||||||
> make docker
|
> make docker
|
||||||
> docker compose -f packages/meta-gateway/dev/docker-compose.yml \
|
> docker compose -f packages/meta-gateway/dev/docker-compose.gateway.yml \
|
||||||
> up -d --force-recreate metagateway-generic-core
|
> up -d --force-recreate metagateway-hub-metacore
|
||||||
> docker compose -f packages/meta-gateway/dev/docker-compose.torznab.yml \
|
|
||||||
> up -d --force-recreate metagateway-torznab-core
|
|
||||||
> docker compose -f packages/meta-share/dev/docker-compose.yml \
|
> docker compose -f packages/meta-share/dev/docker-compose.yml \
|
||||||
> --profile share-10node up -d --force-recreate metashare-core metashare-core-test
|
> --profile share-10node up -d --force-recreate metashare-core metashare-core-test
|
||||||
|
> docker compose -f packages/meta-read/dev/docker-compose.yml \
|
||||||
|
> up -d --force-recreate metaread-dev-core
|
||||||
|
> docker compose -f packages/meta-listen/dev/docker-compose.yml \
|
||||||
|
> up -d --force-recreate metalisten-dev-core
|
||||||
> ```
|
> ```
|
||||||
>
|
>
|
||||||
> The symptom of a missed recreate is missing API fields on `/urls`
|
> The symptom of a missed recreate is missing API fields on `/urls`
|
||||||
@@ -109,7 +117,7 @@ sub-stacks can't see a freshly-built image by name.
|
|||||||
>
|
>
|
||||||
> ```bash
|
> ```bash
|
||||||
> docker inspect meta-core:local --format '{{.Id}}' # fresh SHA
|
> docker inspect meta-core:local --format '{{.Id}}' # fresh SHA
|
||||||
> docker ps --filter ancestor=meta-core:local --format '{{.Names}}'
|
> docker ps --filter ancestor=meta-core:local --format '{{.Names}}' # repeat for :latest
|
||||||
> # any container whose Image SHA doesn't match the fresh SHA needs --force-recreate.
|
> # any container whose Image SHA doesn't match the fresh SHA needs --force-recreate.
|
||||||
> ```
|
> ```
|
||||||
|
|
||||||
@@ -126,36 +134,39 @@ HTTP port (no auth, bypasses Caddy) is `http://localhost:18083`.
|
|||||||
|
|
||||||
## Configuration
|
## Configuration
|
||||||
|
|
||||||
Environment variables consumed by `internal/config/config.go`:
|
Environment variables (from `internal/config/config.go` unless noted):
|
||||||
|
|
||||||
| Variable | Default | Description |
|
| Variable | Default | Description |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `META_CORE_PATH` | `/meta-core` | Shared volume root (locks, db, services, mounts, cache). |
|
| `META_CORE_PATH` | `/meta-core` | meta-core's own volume root (lock, Redis data, identity, mounts, watchers, cache). No other service mounts it. |
|
||||||
| `FILES_PATH` | `/files` | Files volume root. |
|
| `FILES_PATH` | `/files` | Files volume root. |
|
||||||
| `SERVICE_NAME` | `meta-core` | Identity in the service registry. |
|
| `SERVICE_NAME` | `meta-core` | Name in the beacon v2 advertise. |
|
||||||
| `SERVICE_VERSION` | `1.0.0` | Reported via `/status` and registry. |
|
| `SERVICE_VERSION` | `1.0.0` | Reported via `/status`. |
|
||||||
| `API_PORT` | `8180` | External port baked into the leader-info `baseUrl`. |
|
| `API_PORT` | `8180` | External port baked into the constructed `baseUrl` when `BASE_URL` is unset. |
|
||||||
| `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` if set (used for HTTPS perimeter). |
|
| `BASE_URL` | _empty_ | Overrides the constructed `baseUrl` (the Caddy/HTTPS perimeter URL). |
|
||||||
|
| `META_CORE_PUBLIC_URL` | _empty_ | Browser-facing nav URL in the announce; wins over `BASE_URL` (e.g. a debug-direct port with no Caddy). |
|
||||||
| `REDIS_PORT` | `6379` | Local Redis port. |
|
| `REDIS_PORT` | `6379` | Local Redis port. |
|
||||||
| `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. |
|
| `META_CORE_HTTP_PORT` | `9000` | Go HTTP+SSE+WebDAV port. |
|
||||||
| `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. |
|
| `META_CORE_HTTP_HOST` | `0.0.0.0` | HTTP bind. |
|
||||||
| `HEALTH_CHECK_INTERVAL_MS` | `5000` | Internal health-loop cadence. |
|
| `ENABLE_FILE_WATCHER` | `true` | Disable to suppress the in-process watcher entirely (`POST /api/files/register` stays available). |
|
||||||
| `HEARTBEAT_INTERVAL_MS` | `30000` | Service registry heartbeat. |
|
| `ENABLE_UDP_DISCOVERY` | `true` | beacon v2 advertise + listen. |
|
||||||
| `STALE_THRESHOLD_MS` | `60000` | Age past which a registry entry is marked stale. |
|
| `BEACON_GROUP` / `BEACON_PORT` | `239.255.99.1` / `9099` | Multicast group and UDP port. |
|
||||||
| `CLEANUP_INTERVAL_MS` | `600000` | Dead-service cleaner cadence. |
|
| `BEACON_INTERVAL_MS` | `10000` | Unsolicited advertise interval. |
|
||||||
| `DEAD_SERVICE_THRESHOLD_MS` | `600000` | Age past which a stale entry is deleted. |
|
| `META_CORE_INDEX_EXCLUDE_PREFIXES` | `categories/newznab/` | Comma-separated field prefixes left out of the in-memory search index; set empty to index everything (`internal/storage/client.go`). |
|
||||||
| `WATCH_INTERVAL_MS` | `1000` | Watcher poll cadence. |
|
| `ELECTION_RETRY_SECS` | `5` | flock retry interval (read by `docker/leader-election.sh`, not the Go binary). |
|
||||||
| `DEBOUNCE_MS` | `30000` | File-event debounce window. |
|
|
||||||
| `ENABLE_FILE_WATCHER` | `true` | Disable to suppress the in-process watcher entirely. |
|
|
||||||
|
|
||||||
Note: there is no `META_CORE_REDIS_PORT`, no `META_CORE_SERVICE_NAME`, and
|
`HEALTH_CHECK_INTERVAL_MS`, `HEARTBEAT_INTERVAL_MS`, `STALE_THRESHOLD_MS`,
|
||||||
no `META_CORE_LEADER_TIMEOUT`. The Redis URL is discovered locally; service
|
`CLEANUP_INTERVAL_MS`, `DEAD_SERVICE_THRESHOLD_MS`, `WATCH_INTERVAL_MS` and
|
||||||
identity uses `SERVICE_NAME`; leader timeout is governed by the bash flock
|
`DEBOUNCE_MS` are still parsed by `internal/config/config.go` but nothing reads
|
||||||
retry interval (`ELECTION_RETRY_SECS`, default `5`).
|
them (leftovers of the file-based registry); setting them has no effect. There
|
||||||
|
is no `META_CORE_REDIS_PORT` / `META_CORE_SERVICE_NAME` /
|
||||||
|
`META_CORE_LEADER_TIMEOUT`, and no `WATCH_PATHS` — watched folders are watcher
|
||||||
|
configs created via `/api/watchers`.
|
||||||
|
|
||||||
## API reference
|
## API reference
|
||||||
|
|
||||||
All endpoints below are served by `internal/api/server.go`. Examples assume
|
All endpoints below are registered in `internal/api/server.go` (mounts,
|
||||||
|
watcher and watchers routes in their own packages' `handlers.go`). Examples assume
|
||||||
the in-container Go port `:9000`; from the host use
|
the in-container Go port `:9000`; from the host use
|
||||||
`https://metacore-dev.localhost:8083` (Caddy) or `http://localhost:18083`
|
`https://metacore-dev.localhost:8083` (Caddy) or `http://localhost:18083`
|
||||||
(debug-direct).
|
(debug-direct).
|
||||||
@@ -167,14 +178,14 @@ GET /health # storage + role
|
|||||||
GET /status # version, uptime, etc.
|
GET /status # version, uptime, etc.
|
||||||
GET /leader # current leader info (hostname, PID, timestamps)
|
GET /leader # current leader info (hostname, PID, timestamps)
|
||||||
GET /urls # baseUrl, apiUrl, webdavUrl, webdavUrlInternal (redisUrl is intentionally empty)
|
GET /urls # baseUrl, apiUrl, webdavUrl, webdavUrlInternal (redisUrl is intentionally empty)
|
||||||
GET /neighbors # neighbours heard over UDP (meta-discovery v1)
|
GET /neighbors # neighbours heard over UDP (beacon v2)
|
||||||
GET /api/neighbors # same, under /api
|
GET /api/neighbors # same, under /api
|
||||||
GET /api/stats # dashboard counters (see below)
|
GET /api/stats # dashboard counters (see below)
|
||||||
```
|
```
|
||||||
|
|
||||||
`/services`, `/api/services`, `/services/{name}` and `/services/cleanup/stats`
|
`/services`, `/api/services`, `/services/{name}` and `/services/cleanup/stats`
|
||||||
are **gone** — the registration-and-heartbeat registry they served was replaced
|
are **gone** — the registration-and-heartbeat registry they served was replaced
|
||||||
by meta-discovery v1 (`/neighbors`). nginx answers an unknown non-`/api` path
|
by beacon v2 (`/neighbors`). nginx answers an unknown non-`/api` path
|
||||||
with the dashboard's `index.html`, so a stale caller gets `200 text/html` rather
|
with the dashboard's `index.html`, so a stale caller gets `200 text/html` rather
|
||||||
than a 404; that is what made the dashboard fail with
|
than a 404; that is what made the dashboard fail with
|
||||||
`SyntaxError: Unexpected token '<'` until 1.0.21.
|
`SyntaxError: Unexpected token '<'` until 1.0.21.
|
||||||
@@ -229,6 +240,8 @@ GET /api/meta/{cid} # resolve CID → metadata document
|
|||||||
GET /api/file/{cid}/info # CID resolution metadata
|
GET /api/file/{cid}/info # CID resolution metadata
|
||||||
GET /file/{cid} # serve file bytes (range-aware)
|
GET /file/{cid} # serve file bytes (range-aware)
|
||||||
POST /file/cid # compute CID for an uploaded file
|
POST /file/cid # compute CID for an uploaded file
|
||||||
|
POST /api/files/register # mint + alias one file under /files/plugin by midhash
|
||||||
|
# (the watcher's path for one file; works with the watcher OFF)
|
||||||
HEAD /data/{hash} # existence + size
|
HEAD /data/{hash} # existence + size
|
||||||
GET /data/{hash}/path # absolute path on /files
|
GET /data/{hash}/path # absolute path on /files
|
||||||
```
|
```
|
||||||
@@ -280,6 +293,44 @@ signalling).
|
|||||||
```bash
|
```bash
|
||||||
GET /api/events/files # file:events stream (watcher events)
|
GET /api/events/files # file:events stream (watcher events)
|
||||||
GET /api/events/meta # meta:events stream (metadata mutations)
|
GET /api/events/meta # meta:events stream (metadata mutations)
|
||||||
|
GET /api/events/poll # DEPRECATED long-poll over file:events — use SSE
|
||||||
|
```
|
||||||
|
|
||||||
|
### Identity (signing keystore)
|
||||||
|
|
||||||
|
Reached through meta-search's identity proxy (and directly by meta-gateway,
|
||||||
|
meta-read, meta-listen) for record signing, UDL writes and keypair sign-in.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
GET /api/identity # status (never the private key)
|
||||||
|
GET /api/identity/accounts
|
||||||
|
POST /api/identity/generate # the only call that returns a new private key
|
||||||
|
POST /api/identity/import
|
||||||
|
POST /api/identity/challenge # one-shot challenge for proof-of-possession
|
||||||
|
POST /api/identity/reveal # needs proof-of-possession
|
||||||
|
DELETE /api/identity # needs proof-of-possession; reports purged UDL counts
|
||||||
|
POST /api/identity/sign
|
||||||
|
POST /api/identity/sign-batch
|
||||||
|
GET /api/identity/aead-key
|
||||||
|
```
|
||||||
|
|
||||||
|
### User Data Layer
|
||||||
|
|
||||||
|
Records arrive signed (private-tier ones encrypted) as opaque base64 CBOR;
|
||||||
|
writes are version-gated (stale → `409 stale_version`, or `accepted:false` per
|
||||||
|
cell in a batch). Redis model in `internal/storage/udl.go`.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
GET /api/udl/record?uid&cid&key
|
||||||
|
PUT /api/udl/record
|
||||||
|
PUT /api/udl/records # batch
|
||||||
|
GET /api/udl/user/{uid}/key/{key}
|
||||||
|
GET /api/udl/user/{uid}/cid/{cid}
|
||||||
|
GET /api/udl/user/{uid}/all
|
||||||
|
GET /api/udl/user/{uid}/count
|
||||||
|
GET /api/udl/cid/{cid}/users
|
||||||
|
GET /api/udl/cid/{cid}/aggregate # meta-watch Aggregate wire shape
|
||||||
|
GET /api/udl/users/stats
|
||||||
```
|
```
|
||||||
|
|
||||||
### Mounts
|
### Mounts
|
||||||
@@ -319,6 +370,9 @@ return a deprecation pointer at the `/api/watchers/*` routes
|
|||||||
|
|
||||||
```bash
|
```bash
|
||||||
POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries
|
POST /api/admin/migrate-dual-roots # reunify stranded midhash-rooted entries
|
||||||
|
POST /api/admin/migrate-domain-screen # domain film|tv → screen, workForm backfill (idempotent)
|
||||||
|
POST /api/admin/migrate-user-cids # backfill udl:idx:user:<uid>:cids; until run, pre-index
|
||||||
|
# profiles export as EMPTY. Returns {fixed}
|
||||||
POST /api/admin/sweep-literature-echoes[?apply=true]
|
POST /api/admin/sweep-literature-echoes[?apply=true]
|
||||||
# strip query echoes off literature cards: anilistid on site cards,
|
# strip query echoes off literature cards: anilistid on site cards,
|
||||||
# workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card.
|
# workcid/chapterNumber/volumeNumber/chapterStart/chapterEnd on any card.
|
||||||
@@ -329,32 +383,29 @@ GET/PUT/DELETE /webdav/... # mounted on /files
|
|||||||
No `/metrics` endpoint is exposed; observability is via container logs and
|
No `/metrics` endpoint is exposed; observability is via container logs and
|
||||||
the `/health` / `/status` JSON.
|
the `/health` / `/status` JSON.
|
||||||
|
|
||||||
## Leader gate
|
## Redis-volume mutex
|
||||||
|
|
||||||
Leader election is a bash flock loop in `docker/leader-election.sh`, not a
|
`docker/leader-election.sh` is a bash flock loop, not a Go-level state
|
||||||
Go-level state machine. When `flock` succeeds on `kv-leader.lock`, the
|
machine. When `flock` succeeds on `/meta-core/locks/kv-leader.lock`, the
|
||||||
script writes the leader API URL to `kv-leader.info` and `exec`s
|
script `exec`s supervisord; a second container mounting the same volume
|
||||||
supervisord. When the process tree dies, the kernel releases the lock and
|
retries every `ELECTION_RETRY_SECS` and stays dormant. When the process tree
|
||||||
any standby instance can take over. There is no follower path inside the
|
dies, the kernel releases the lock. There is no follower path inside the Go
|
||||||
Go binary — there is no `OnBecomeLeader` / `OnBecomeFollower` callback API.
|
binary.
|
||||||
|
|
||||||
`kv-leader.info` is a single plaintext line containing the API base URL:
|
The lock is a **mutex, not service discovery** — it only stops two Redis
|
||||||
|
writers landing on one RDB/AOF. Nothing reads it, and the `kv-leader.info`
|
||||||
```
|
file it used to publish is gone: siblings locate meta-core over UDP
|
||||||
http://10.0.1.50:9000
|
(see [beacon-v2.md](../../docs/project-architecture/beacon-v2.md)).
|
||||||
```
|
`internal/leader` keeps its name for history, but only builds the `/urls`
|
||||||
|
payload (`baseUrl`, `apiUrl`, `webdavUrl`, `webdavUrlInternal`) that `/urls`
|
||||||
Other services read this file, then call `GET /urls` for full discovery.
|
and the UDP announce share.
|
||||||
|
|
||||||
The dev compose runs a single `metacore-app`; the flock primitive is
|
|
||||||
retained as a correctness anchor (one process owns Redis and the
|
|
||||||
writable state under `/meta-core`), not as live failover.
|
|
||||||
|
|
||||||
## Metadata storage shape (one-line summary)
|
## Metadata storage shape (one-line summary)
|
||||||
|
|
||||||
Roots are opaque UUIDv7 strings; every property is a separate Redis
|
Roots are opaque UUIDv7 strings; every property is a separate Redis
|
||||||
STRING at `file:<uuid>/<property>`; every known CID for a file is a
|
STRING at `file:<uuid>/<property>`; every known CID is a key-set member
|
||||||
reverse-index entry `cid:<algorithm>:<value> → <uuid>`. Full details in
|
`file:<uuid>/cids/<bareCid>` plus a reverse-index entry `cid:<bareCid> → <uuid>`
|
||||||
|
(the canonical CID is derived by rank on read, never stored). Full details in
|
||||||
[`docs/metadata-storage-structure.md`](docs/metadata-storage-structure.md)
|
[`docs/metadata-storage-structure.md`](docs/metadata-storage-structure.md)
|
||||||
and design rationale in [`docs/uuid-rooted-metadata.md`](docs/uuid-rooted-metadata.md).
|
and design rationale in [`docs/uuid-rooted-metadata.md`](docs/uuid-rooted-metadata.md).
|
||||||
The authoritative registry of field semantics and value formats lives in
|
The authoritative registry of field semantics and value formats lives in
|
||||||
@@ -364,14 +415,15 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
|
|||||||
|
|
||||||
| Package | Purpose |
|
| Package | Purpose |
|
||||||
|---------|---------|
|
|---------|---------|
|
||||||
| `cmd/meta-core` | Entry point; Redis connect + schema sentinel + service wiring. |
|
| `cmd/meta-core` | Entry point; Redis connect + schema sentinel + UDP discovery + service wiring. |
|
||||||
| `internal/config` | Env-driven configuration, path helpers. |
|
| `internal/config` | Env-driven configuration, path helpers. |
|
||||||
| `internal/leader` | `LeaderInfoProvider` (builds the leader info from local hostname/IP/config; no election logic). |
|
| `internal/leader` | `LeaderInfoProvider` — builds the `/urls` payload from hostname/IP/config; no election logic. |
|
||||||
| `internal/storage` | Redis wrapper (`client.go`), UUIDv7 minting (`uuid.go`), CID reverse index (`cid_resolution.go`), schema sentinel (`schema_version.go`), dual-root migration (`dual_root_migration.go`). |
|
| `internal/meshdisco` | beacon v2 node (advertise, probe, neighbour map) — the Go port of the spec. |
|
||||||
| `internal/cid` | CID parsing + algorithm ranking for canonical-CID selection. |
|
| `internal/storage` | Redis wrapper (`client.go`), UUIDv7 minting, CID reverse index (`cid_resolution.go`), schema sentinel, field indexes + search-index bulk reads, UDL store (`udl.go`, `udl_admin.go`), stats, one-shot migrations/sweeps. |
|
||||||
| `internal/discovery` | Service registry + dead-service cleaner. |
|
| `internal/cid` | CID parsing + rank ladder for canonical-CID selection (`cid-rank-vectors.json`, guarded by the meta-root `scripts/check-cid-vectors.sh`). |
|
||||||
|
| `internal/identity` | Multi-account secp256k1 keystore, challenge/proof-of-possession, signature verification. |
|
||||||
| `internal/events` | Keyspace-notification → `meta:events` stream publisher (`meta_publisher.go`). |
|
| `internal/events` | Keyspace-notification → `meta:events` stream publisher (`meta_publisher.go`). |
|
||||||
| `internal/api` | HTTP server + all handlers + SSE event endpoints. |
|
| `internal/api` | HTTP server + all handlers + SSE event endpoints + in-memory search index + `/api/stats`. |
|
||||||
| `internal/schema` | Live per-field schema indexer (consumes `meta:events`). |
|
| `internal/schema` | Live per-field schema indexer (consumes `meta:events`). |
|
||||||
| `internal/snapshot` | Snapshot export / import / wipe. |
|
| `internal/snapshot` | Snapshot export / import / wipe. |
|
||||||
| `internal/watcher` | File system scanner, MidHash256 computation, dispatcher, state registry. |
|
| `internal/watcher` | File system scanner, MidHash256 computation, dispatcher, state registry. |
|
||||||
@@ -386,10 +438,12 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
|
|||||||
2. Build LeaderInfoProvider (no election — the bash gate already won)
|
2. Build LeaderInfoProvider (no election — the bash gate already won)
|
||||||
3. Connect to local Redis (retry up to 30× / 1s)
|
3. Connect to local Redis (retry up to 30× / 1s)
|
||||||
4. EnsureSchemaVersion — abort if the existing Redis layout is stale
|
4. EnsureSchemaVersion — abort if the existing Redis layout is stale
|
||||||
5. Start service discovery + dead-service cleaner
|
5. Start beacon v2 node (unless ENABLE_UDP_DISCOVERY=false)
|
||||||
6. Construct API server (initialises watcher, mounts, WebDAV)
|
6. Construct API server (identity keystore migration, watcher, watchers,
|
||||||
7. Start API server; on first connect, start MetaPublisher + schema Indexer
|
mounts, WebDAV)
|
||||||
and republish existing metadata onto meta:events
|
7. Start API server + pollers + search-index refresh loop; start
|
||||||
|
MetaPublisher + schema Indexer, then warm field indexes and republish
|
||||||
|
existing metadata onto meta:events in the background
|
||||||
8. Wait for SIGINT/SIGTERM, shutdown in reverse order
|
8. Wait for SIGINT/SIGTERM, shutdown in reverse order
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -400,11 +454,13 @@ the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md).
|
|||||||
| `github.com/gorilla/mux` | HTTP router |
|
| `github.com/gorilla/mux` | HTTP router |
|
||||||
| `github.com/redis/go-redis/v9` | Redis client |
|
| `github.com/redis/go-redis/v9` | Redis client |
|
||||||
| `github.com/google/uuid` | UUIDv7 minting |
|
| `github.com/google/uuid` | UUIDv7 minting |
|
||||||
|
| `github.com/decred/dcrd/dcrec/secp256k1/v4`, `golang.org/x/crypto`, `github.com/mr-tron/base58` | Identity keys, signing, AEAD key derivation |
|
||||||
| `golang.org/x/net` | WebDAV support |
|
| `golang.org/x/net` | WebDAV support |
|
||||||
| Go stdlib | context, net/http, os, sync, syscall |
|
| `github.com/alicebob/miniredis/v2` | In-memory Redis for tests |
|
||||||
|
|
||||||
Runtime requirements: Go 1.21+ at build time; the container ships Redis,
|
Runtime requirements: Go 1.21+ at build time; the container ships Redis,
|
||||||
rclone, nginx, and supervisord alongside the binary.
|
rclone, nginx, and supervisord alongside the binary, plus the React dashboard
|
||||||
|
and metadata editor (built from `dashboard/` and `editor/` in the Dockerfile).
|
||||||
|
|
||||||
## License
|
## License
|
||||||
|
|
||||||
|
|||||||
+20
-9
@@ -1,6 +1,7 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"os"
|
"os"
|
||||||
@@ -60,34 +61,44 @@ func main() {
|
|||||||
}
|
}
|
||||||
log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion)
|
log.Printf("[meta-core] Redis schema version OK (v%d)", storage.SchemaVersion)
|
||||||
|
|
||||||
// UDP service discovery (meta-discovery v1). Runs alongside the file
|
// Beacon v2 (docs/project-architecture/beacon-v2.md): advertise meta-core
|
||||||
// registry above for now — nothing consumes the wire yet, so this cannot
|
// as metamesh.core — carrying the /urls block services locate us by — plus
|
||||||
// regress the existing path. The announce payload is rebuilt from
|
// its own metamesh.service/<name> nav entry. The payload is rebuilt from
|
||||||
// leaderProvider on every tick, which is what keeps it and GET /urls from
|
// leaderProvider on every tick, which keeps it and GET /urls from drifting.
|
||||||
// drifting apart.
|
|
||||||
var mesh *meshdisco.Node
|
var mesh *meshdisco.Node
|
||||||
if cfg.EnableUDPDiscovery {
|
if cfg.EnableUDPDiscovery {
|
||||||
hostname, _ := os.Hostname()
|
hostname, _ := os.Hostname()
|
||||||
mesh = meshdisco.New(meshdisco.Config{
|
mesh = meshdisco.New(meshdisco.Config{
|
||||||
Name: cfg.ServiceName,
|
Name: cfg.ServiceName,
|
||||||
Instance: hostname,
|
Instance: hostname,
|
||||||
Role: meshdisco.RoleCore,
|
|
||||||
Version: Version,
|
Version: Version,
|
||||||
Group: cfg.DiscoveryGroup,
|
Group: cfg.DiscoveryGroup,
|
||||||
Port: cfg.DiscoveryPort,
|
Port: cfg.DiscoveryPort,
|
||||||
Interval: time.Duration(cfg.DiscoveryIntervalMS) * time.Millisecond,
|
Interval: time.Duration(cfg.DiscoveryIntervalMS) * time.Millisecond,
|
||||||
Payload: func() (string, string, *meshdisco.URLs) {
|
Payload: func() (string, []meshdisco.Resource) {
|
||||||
|
service := meshdisco.Resource{
|
||||||
|
ID: "service",
|
||||||
|
Caps: []string{meshdisco.CapServicePrefix + cfg.ServiceName},
|
||||||
|
Endpoints: map[string]string{"ui": coreBaseURL(cfg)},
|
||||||
|
}
|
||||||
info := leaderProvider.LeaderInfo()
|
info := leaderProvider.LeaderInfo()
|
||||||
if info == nil {
|
if info == nil {
|
||||||
return "", "starting", nil
|
return "starting", []meshdisco.Resource{service}
|
||||||
}
|
}
|
||||||
return coreBaseURL(cfg), "running", &meshdisco.URLs{
|
data, _ := json.Marshal(map[string]meshdisco.URLs{"urls": {
|
||||||
Hostname: info.Hostname,
|
Hostname: info.Hostname,
|
||||||
BaseUrl: info.BaseUrl,
|
BaseUrl: info.BaseUrl,
|
||||||
ApiUrl: info.ApiUrl,
|
ApiUrl: info.ApiUrl,
|
||||||
WebdavUrl: info.WebdavUrl,
|
WebdavUrl: info.WebdavUrl,
|
||||||
WebdavUrlInternal: info.WebdavUrlInternal,
|
WebdavUrlInternal: info.WebdavUrlInternal,
|
||||||
|
}})
|
||||||
|
core := meshdisco.Resource{
|
||||||
|
ID: "core",
|
||||||
|
Caps: []string{meshdisco.CapCore},
|
||||||
|
Endpoints: map[string]string{"http": info.ApiUrl},
|
||||||
|
Data: data,
|
||||||
}
|
}
|
||||||
|
return "running", []meshdisco.Resource{core, service}
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
if err := mesh.Start(); err != nil {
|
if err := mesh.Start(); err != nil {
|
||||||
|
|||||||
@@ -238,6 +238,10 @@ class MetaServiceMenu extends HTMLElement {
|
|||||||
for (const n of this._neighbors) {
|
for (const n of this._neighbors) {
|
||||||
const name = n.name || "";
|
const name = n.name || "";
|
||||||
if (!name || seen.has(name)) continue;
|
if (!name || seen.has(name)) continue;
|
||||||
|
// Beacon v2 neighbours include plugins; the nav lists services only.
|
||||||
|
// (A row without `caps` comes from a pre-v2 backend — keep it.)
|
||||||
|
const isService = (c) => String(c).startsWith("metamesh.service/");
|
||||||
|
if (Array.isArray(n.caps) && !n.caps.some(isService)) continue;
|
||||||
seen.add(name);
|
seen.add(name);
|
||||||
rows.push(n);
|
rows.push(n);
|
||||||
}
|
}
|
||||||
@@ -275,7 +279,7 @@ class MetaServiceMenu extends HTMLElement {
|
|||||||
el.href = url;
|
el.href = url;
|
||||||
el.setAttribute("role", "menuitem");
|
el.setAttribute("role", "menuitem");
|
||||||
}
|
}
|
||||||
if (!isActive && !url) el.title = "No reachable URL announced";
|
if (!isActive && !url) el.title = "No reachable URL advertised";
|
||||||
|
|
||||||
const icon = document.createElement("span");
|
const icon = document.createElement("span");
|
||||||
icon.className = "icon";
|
icon.className = "icon";
|
||||||
@@ -285,7 +289,8 @@ class MetaServiceMenu extends HTMLElement {
|
|||||||
label.textContent = titleCase(n.name);
|
label.textContent = titleCase(n.name);
|
||||||
|
|
||||||
el.append(icon, label);
|
el.append(icon, label);
|
||||||
if (n.status === "running" || isActive) {
|
// Beacon v2 omits `status` when running.
|
||||||
|
if ((n.status || "running") === "running" || isActive) {
|
||||||
const dot = document.createElement("span");
|
const dot = document.createElement("span");
|
||||||
dot.className = "dot";
|
dot.className = "dot";
|
||||||
dot.title = n.status || "running";
|
dot.title = n.status || "running";
|
||||||
|
|||||||
@@ -18,5 +18,5 @@ require (
|
|||||||
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
github.com/cespare/xxhash/v2 v2.2.0 // indirect
|
||||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||||
github.com/yuin/gopher-lua v1.1.0 // indirect
|
github.com/yuin/gopher-lua v1.1.0 // indirect
|
||||||
golang.org/x/sys v0.28.0 // indirect
|
golang.org/x/sys v0.28.0
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -550,11 +550,11 @@ func (s *Server) handleGetFileByCID(w http.ResponseWriter, r *http.Request) {
|
|||||||
|
|
||||||
// handleListNeighbors handles GET /neighbors and GET /api/neighbors.
|
// handleListNeighbors handles GET /neighbors and GET /api/neighbors.
|
||||||
//
|
//
|
||||||
// meta-discovery v1. The list is built from UDP announces held in memory, not
|
// Beacon v2 (docs/project-architecture/beacon-v2.md). The list is this node's
|
||||||
// from files under /meta-core/services, so it works across stacks that share a
|
// in-memory view of the UDP group, so it works across stacks that share a
|
||||||
// network but no volume. Instances of the same service are collapsed to the
|
// network but no volume. Instances of one name collapse to the most recent
|
||||||
// most recent, which preserves the one-row-per-service shape the old
|
// (one row per service, the nav menu's shape); `?all=1` returns every
|
||||||
// /api/services had. `?all=1` returns every instance, for debugging.
|
// instance, `?cap=<pattern>` keeps only nodes with a matching capability.
|
||||||
func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
|
func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
|
||||||
response := map[string]interface{}{
|
response := map[string]interface{}{
|
||||||
"current": s.config.ServiceName,
|
"current": s.config.ServiceName,
|
||||||
@@ -576,6 +576,15 @@ func (s *Server) handleListNeighbors(w http.ResponseWriter, r *http.Request) {
|
|||||||
} else {
|
} else {
|
||||||
list = s.mesh.NeighborsByName()
|
list = s.mesh.NeighborsByName()
|
||||||
}
|
}
|
||||||
|
if pattern := r.URL.Query().Get("cap"); pattern != "" {
|
||||||
|
kept := list[:0]
|
||||||
|
for _, nb := range list {
|
||||||
|
if nb.Matches(pattern) {
|
||||||
|
kept = append(kept, nb)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
list = kept
|
||||||
|
}
|
||||||
|
|
||||||
response["enabled"] = true
|
response["enabled"] = true
|
||||||
response["neighbors"] = list
|
response["neighbors"] = list
|
||||||
|
|||||||
@@ -43,7 +43,7 @@ type Server struct {
|
|||||||
// challenges backs proof-of-possession on reveal/delete. In-process, so a
|
// challenges backs proof-of-possession on reveal/delete. In-process, so a
|
||||||
// restart invalidates the ones in flight — see identity/challenge.go.
|
// restart invalidates the ones in flight — see identity/challenge.go.
|
||||||
challenges *identity.ChallengeStore
|
challenges *identity.ChallengeStore
|
||||||
// mesh is the UDP discovery node (meta-discovery v1). nil when discovery
|
// mesh is the UDP discovery node (beacon v2). nil when discovery
|
||||||
// is disabled or failed to bind, in which case /api/neighbors reports an
|
// is disabled or failed to bind, in which case /api/neighbors reports an
|
||||||
// empty list rather than erroring — a dashboard nav is decoration and
|
// empty list rather than erroring — a dashboard nav is decoration and
|
||||||
// must not take the API down with it.
|
// must not take the API down with it.
|
||||||
@@ -291,7 +291,7 @@ func (s *Server) setupRoutes() {
|
|||||||
s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET")
|
s.router.HandleFunc("/api/events/files", s.handleEventsFiles).Methods("GET")
|
||||||
s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET")
|
s.router.HandleFunc("/api/events/meta", s.handleEventsMeta).Methods("GET")
|
||||||
|
|
||||||
// meta-discovery v1: neighbours heard over UDP. Supersedes /api/services,
|
// beacon v2: neighbours heard over UDP. Supersedes /api/services,
|
||||||
// which is still served above while consumers migrate.
|
// which is still served above while consumers migrate.
|
||||||
s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET")
|
s.router.HandleFunc("/neighbors", s.handleListNeighbors).Methods("GET")
|
||||||
s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET")
|
s.router.HandleFunc("/api/neighbors", s.handleListNeighbors).Methods("GET")
|
||||||
|
|||||||
+11
-11
@@ -41,13 +41,13 @@ type Config struct {
|
|||||||
// Mount configuration
|
// Mount configuration
|
||||||
MountsDir string // Path to mounts configuration (default: /meta-core/mounts)
|
MountsDir string // Path to mounts configuration (default: /meta-core/mounts)
|
||||||
|
|
||||||
// UDP service discovery (meta-discovery v1). Replaces the file-based
|
// UDP discovery (beacon v2). Replaces the file-based registry that used
|
||||||
// registry that used to live under /meta-core/services; see
|
// to live under /meta-core/services; see
|
||||||
// docs/project-architecture/service-discovery.md.
|
// docs/project-architecture/beacon-v2.md.
|
||||||
EnableUDPDiscovery bool // Announce + listen on the multicast group (default: true)
|
EnableUDPDiscovery bool // Advertise + listen on the multicast group (default: true)
|
||||||
DiscoveryGroup string // IPv4 multicast group (default: 239.255.77.1)
|
DiscoveryGroup string // BEACON_GROUP: IPv4 multicast group (default: 239.255.99.1)
|
||||||
DiscoveryPort int // UDP port (default: 9399)
|
DiscoveryPort int // BEACON_PORT: UDP port (default: 9099)
|
||||||
DiscoveryIntervalMS int // Unsolicited announce interval in ms (default: 10000)
|
DiscoveryIntervalMS int // BEACON_INTERVAL_MS: unsolicited advertise interval (default: 10000)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Load creates a Config from environment variables
|
// Load creates a Config from environment variables
|
||||||
@@ -72,9 +72,9 @@ func Load() *Config {
|
|||||||
DebounceMS: getEnvInt("DEBOUNCE_MS", 30000),
|
DebounceMS: getEnvInt("DEBOUNCE_MS", 30000),
|
||||||
EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true),
|
EnableFileWatcher: getEnvBool("ENABLE_FILE_WATCHER", true),
|
||||||
EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true),
|
EnableUDPDiscovery: getEnvBool("ENABLE_UDP_DISCOVERY", true),
|
||||||
DiscoveryGroup: getEnv("DISCOVERY_GROUP", "239.255.77.1"),
|
DiscoveryGroup: getEnv("BEACON_GROUP", "239.255.99.1"),
|
||||||
DiscoveryPort: getEnvInt("DISCOVERY_PORT", 9399),
|
DiscoveryPort: getEnvInt("BEACON_PORT", 9099),
|
||||||
DiscoveryIntervalMS: getEnvInt("DISCOVERY_INTERVAL_MS", 10000),
|
DiscoveryIntervalMS: getEnvInt("BEACON_INTERVAL_MS", 10000),
|
||||||
}
|
}
|
||||||
|
|
||||||
// Set mounts directory
|
// Set mounts directory
|
||||||
@@ -85,7 +85,7 @@ func Load() *Config {
|
|||||||
|
|
||||||
// LockFilePath returns the path to the Redis-volume mutex. Held by
|
// LockFilePath returns the path to the Redis-volume mutex. Held by
|
||||||
// docker/leader-election.sh; nothing reads it, and it is NOT service
|
// docker/leader-election.sh; nothing reads it, and it is NOT service
|
||||||
// discovery — see docs/project-architecture/service-discovery.md.
|
// discovery — see docs/project-architecture/beacon-v2.md.
|
||||||
func (c *Config) LockFilePath() string {
|
func (c *Config) LockFilePath() string {
|
||||||
return c.MetaCorePath + "/locks/kv-leader.lock"
|
return c.MetaCorePath + "/locks/kv-leader.lock"
|
||||||
}
|
}
|
||||||
|
|||||||
+129
-124
@@ -13,24 +13,24 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"golang.org/x/net/ipv4"
|
"golang.org/x/net/ipv4"
|
||||||
|
"golang.org/x/sys/unix"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Config configures a Node. Payload is a function rather than a value so the
|
// Config configures a Node. Payload is a function rather than a value so the
|
||||||
// announce is rebuilt from live state on every tick — that is what keeps the
|
// advertise is rebuilt from live state on every tick — that is what keeps the
|
||||||
// announced URLs and GET /urls from drifting apart.
|
// advertised core URLs and GET /urls from drifting apart.
|
||||||
type Config struct {
|
type Config struct {
|
||||||
Name string
|
Name string
|
||||||
Instance string
|
Instance string
|
||||||
Role string
|
|
||||||
Version string
|
Version string
|
||||||
|
|
||||||
Group string
|
Group string
|
||||||
Port int
|
Port int
|
||||||
Interval time.Duration
|
Interval time.Duration
|
||||||
|
|
||||||
// Payload returns the identity/URL fields for this announce. Name,
|
// Payload returns this node's status ("starting" / "running") and its
|
||||||
// Instance, Role, Version and the envelope are filled in by the Node.
|
// resources. The envelope and NodeInfo are filled in by the Node.
|
||||||
Payload func() (baseURL string, status string, urls *URLs)
|
Payload func() (status string, resources []Resource)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Config) applyDefaults() {
|
func (c *Config) applyDefaults() {
|
||||||
@@ -43,13 +43,17 @@ func (c *Config) applyDefaults() {
|
|||||||
if c.Interval <= 0 {
|
if c.Interval <= 0 {
|
||||||
c.Interval = DefaultInterval
|
c.Interval = DefaultInterval
|
||||||
}
|
}
|
||||||
if c.Role == "" {
|
|
||||||
c.Role = RoleService
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Node is one participant in the mesh: it announces itself, answers probes,
|
type seen struct {
|
||||||
// and keeps a TTL map of everyone else it has heard from.
|
node NodeInfo
|
||||||
|
resources []Resource
|
||||||
|
addr string
|
||||||
|
lastSeen time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// Node is one beacon v2 participant: it advertises its resources, answers
|
||||||
|
// probes, and keeps a TTL map of every other node it has heard from.
|
||||||
type Node struct {
|
type Node struct {
|
||||||
cfg Config
|
cfg Config
|
||||||
group net.IP
|
group net.IP
|
||||||
@@ -60,7 +64,7 @@ type Node struct {
|
|||||||
// ifaces are the multicast-capable interfaces we joined and send on.
|
// ifaces are the multicast-capable interfaces we joined and send on.
|
||||||
// A container attached to two docker networks (metashare-app on `pcs`
|
// A container attached to two docker networks (metashare-app on `pcs`
|
||||||
// and `metamesh-mesh`, metawatch-* on `metawatch` and `metamesh-mesh`)
|
// and `metamesh-mesh`, metawatch-* on `metawatch` and `metamesh-mesh`)
|
||||||
// has two entries here, and MUST be announced on both — sending via the
|
// has two entries here, and MUST be advertised on both — sending via the
|
||||||
// default route alone silently covers only one network.
|
// default route alone silently covers only one network.
|
||||||
ifaces []net.Interface
|
ifaces []net.Interface
|
||||||
|
|
||||||
@@ -69,7 +73,7 @@ type Node struct {
|
|||||||
sendMu sync.Mutex
|
sendMu sync.Mutex
|
||||||
|
|
||||||
mu sync.RWMutex
|
mu sync.RWMutex
|
||||||
neighbors map[string]*Neighbor
|
nodes map[string]*seen
|
||||||
|
|
||||||
stopChan chan struct{}
|
stopChan chan struct{}
|
||||||
stopOnce sync.Once
|
stopOnce sync.Once
|
||||||
@@ -81,26 +85,25 @@ func New(cfg Config) *Node {
|
|||||||
cfg.applyDefaults()
|
cfg.applyDefaults()
|
||||||
return &Node{
|
return &Node{
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
neighbors: make(map[string]*Neighbor),
|
nodes: make(map[string]*seen),
|
||||||
stopChan: make(chan struct{}),
|
stopChan: make(chan struct{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start opens the multicast socket, joins the group on every eligible
|
// Start opens the multicast socket, joins the group on every eligible
|
||||||
// interface, and begins announcing. A join failure on one interface is logged
|
// interface, and begins advertising. A join failure on one interface is logged
|
||||||
// and skipped rather than fatal — matching the ENABLE_MDNS precedent in the
|
// and skipped rather than fatal — a host that blocks multicast degrades
|
||||||
// Rust services, where a host that blocks multicast degrades instead of
|
// instead of failing to boot.
|
||||||
// failing to boot.
|
|
||||||
func (n *Node) Start() error {
|
func (n *Node) Start() error {
|
||||||
n.group = net.ParseIP(n.cfg.Group)
|
n.group = net.ParseIP(n.cfg.Group)
|
||||||
if n.group == nil || n.group.To4() == nil {
|
if n.group == nil || n.group.To4() == nil {
|
||||||
return fmt.Errorf("meshdisco: invalid multicast group %q", n.cfg.Group)
|
return fmt.Errorf("beacon: invalid multicast group %q", n.cfg.Group)
|
||||||
}
|
}
|
||||||
|
|
||||||
lc := net.ListenConfig{Control: reusePort}
|
lc := net.ListenConfig{Control: reusePort}
|
||||||
conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port))
|
conn, err := lc.ListenPacket(context.Background(), "udp4", fmt.Sprintf(":%d", n.cfg.Port))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("meshdisco: listen on :%d: %w", n.cfg.Port, err)
|
return fmt.Errorf("beacon: listen on :%d: %w", n.cfg.Port, err)
|
||||||
}
|
}
|
||||||
n.closer = conn
|
n.closer = conn
|
||||||
|
|
||||||
@@ -108,41 +111,41 @@ func (n *Node) Start() error {
|
|||||||
// TTL 1: link-local only. Discovery never routes off the local segment;
|
// TTL 1: link-local only. Discovery never routes off the local segment;
|
||||||
// the two-host topology keeps two separate meshes by design.
|
// the two-host topology keeps two separate meshes by design.
|
||||||
if err := pc.SetMulticastTTL(1); err != nil {
|
if err := pc.SetMulticastTTL(1); err != nil {
|
||||||
log.Printf("[meshdisco] Warning: could not set multicast TTL: %v", err)
|
log.Printf("[beacon] Warning: could not set multicast TTL: %v", err)
|
||||||
}
|
}
|
||||||
n.pc = pc
|
n.pc = pc
|
||||||
|
|
||||||
n.ifaces = eligibleInterfaces()
|
n.ifaces = eligibleInterfaces()
|
||||||
if len(n.ifaces) == 0 {
|
if len(n.ifaces) == 0 {
|
||||||
log.Println("[meshdisco] Warning: no multicast-capable interfaces found; discovery will be inert")
|
log.Println("[beacon] Warning: no multicast-capable interfaces found; discovery will be inert")
|
||||||
}
|
}
|
||||||
|
|
||||||
group := &net.UDPAddr{IP: n.group}
|
group := &net.UDPAddr{IP: n.group}
|
||||||
joined := 0
|
joined := 0
|
||||||
for i := range n.ifaces {
|
for i := range n.ifaces {
|
||||||
if err := pc.JoinGroup(&n.ifaces[i], group); err != nil {
|
if err := pc.JoinGroup(&n.ifaces[i], group); err != nil {
|
||||||
log.Printf("[meshdisco] Warning: join %s on %s failed: %v", n.cfg.Group, n.ifaces[i].Name, err)
|
log.Printf("[beacon] Warning: join %s on %s failed: %v", n.cfg.Group, n.ifaces[i].Name, err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
joined++
|
joined++
|
||||||
}
|
}
|
||||||
log.Printf("[meshdisco] Listening on %s:%d as %s/%s (joined %d of %d interfaces)",
|
log.Printf("[beacon] Listening on %s:%d as %s/%s (joined %d of %d interfaces)",
|
||||||
n.cfg.Group, n.cfg.Port, n.cfg.Name, n.cfg.Instance, joined, len(n.ifaces))
|
n.cfg.Group, n.cfg.Port, n.cfg.Name, n.cfg.Instance, joined, len(n.ifaces))
|
||||||
|
|
||||||
n.wg.Add(2)
|
n.wg.Add(2)
|
||||||
go n.readLoop()
|
go n.readLoop()
|
||||||
go n.announceLoop()
|
go n.advertiseLoop()
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stop closes the socket and waits for the loops to exit. It sends a final
|
// Stop multicasts a bye — so neighbours drop us immediately instead of
|
||||||
// announce with status "stopping" so neighbours drop us immediately instead of
|
// waiting out the staleness window — then closes the socket and waits for the
|
||||||
// waiting out the staleness window.
|
// loops to exit.
|
||||||
func (n *Node) Stop() error {
|
func (n *Node) Stop() error {
|
||||||
var err error
|
var err error
|
||||||
n.stopOnce.Do(func() {
|
n.stopOnce.Do(func() {
|
||||||
n.sendAnnounce(nil, "stopping")
|
n.broadcast(Message{Proto: Proto, V: Version, Type: TypeBye, Node: n.nodeInfo("")})
|
||||||
close(n.stopChan)
|
close(n.stopChan)
|
||||||
if n.closer != nil {
|
if n.closer != nil {
|
||||||
err = n.closer.Close()
|
err = n.closer.Close()
|
||||||
@@ -152,28 +155,27 @@ func (n *Node) Stop() error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Probe multicasts a discovery message on every interface. Every listener
|
// Probe multicasts a probe on every interface. Every node owning a resource
|
||||||
// replies immediately, so a caller does not have to wait out an announce
|
// matching one of want (every node, when want is empty) replies immediately,
|
||||||
// interval. This is what makes the protocol usable on the boot path.
|
// so a caller does not have to wait out an interval.
|
||||||
func (n *Node) Probe() {
|
func (n *Node) Probe(want ...string) {
|
||||||
n.broadcast(Message{V: ProtocolVersion, Type: TypeDiscovery})
|
n.broadcast(Message{Proto: Proto, V: Version, Type: TypeProbe, From: n.cfg.Instance, Want: want})
|
||||||
}
|
}
|
||||||
|
|
||||||
// Neighbors returns everyone heard from within the staleness window, sorted by
|
// Neighbors returns every node heard from within the staleness window,
|
||||||
// name. Entries older than that are dropped as a side effect — expiry is a
|
// sorted by (name, instance). Expired entries are dropped as a side effect —
|
||||||
// read-time concern, so there is no reaper goroutine and no equivalent of the
|
// expiry is a read-time concern, so there is no reaper goroutine.
|
||||||
// old cleaner.go.
|
|
||||||
func (n *Node) Neighbors() []Neighbor {
|
func (n *Node) Neighbors() []Neighbor {
|
||||||
cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor)
|
cutoff := time.Now().Add(-n.cfg.Interval * LivenessFactor)
|
||||||
|
|
||||||
n.mu.Lock()
|
n.mu.Lock()
|
||||||
out := make([]Neighbor, 0, len(n.neighbors))
|
out := make([]Neighbor, 0, len(n.nodes))
|
||||||
for k, nb := range n.neighbors {
|
for k, s := range n.nodes {
|
||||||
if nb.LastSeen.Before(cutoff) {
|
if s.lastSeen.Before(cutoff) {
|
||||||
delete(n.neighbors, k)
|
delete(n.nodes, k)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
out = append(out, *nb)
|
out = append(out, newNeighbor(s.node, s.resources, s.addr, s.lastSeen))
|
||||||
}
|
}
|
||||||
n.mu.Unlock()
|
n.mu.Unlock()
|
||||||
|
|
||||||
@@ -186,13 +188,12 @@ func (n *Node) Neighbors() []Neighbor {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// NeighborsByName collapses multiple instances of one service down to the most
|
// NeighborsByName collapses multiple instances of one name down to the most
|
||||||
// recently seen, which is the shape the nav menu wants (and what the existing
|
// recently seen — the shape the nav menu wants.
|
||||||
// /api/services dedup contract asserts).
|
|
||||||
func (n *Node) NeighborsByName() []Neighbor {
|
func (n *Node) NeighborsByName() []Neighbor {
|
||||||
best := make(map[string]Neighbor)
|
best := make(map[string]Neighbor)
|
||||||
for _, nb := range n.Neighbors() {
|
for _, nb := range n.Neighbors() {
|
||||||
if cur, ok := best[nb.Name]; !ok || nb.LastSeen.After(cur.LastSeen) {
|
if cur, ok := best[nb.Name]; !ok || nb.LastSeen > cur.LastSeen {
|
||||||
best[nb.Name] = nb
|
best[nb.Name] = nb
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -204,10 +205,27 @@ func (n *Node) NeighborsByName() []Neighbor {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// Self returns this node's own announce, so callers can render themselves in
|
// Self returns this node as a row, so callers can render themselves without
|
||||||
// the menu without waiting to hear their own multicast echo back.
|
// waiting to hear their own multicast echo back.
|
||||||
func (n *Node) Self() Neighbor {
|
func (n *Node) Self() Neighbor {
|
||||||
return Neighbor{Message: n.buildAnnounce("running"), LastSeen: time.Now()}
|
status, resources := n.payload()
|
||||||
|
return newNeighbor(*n.nodeInfo(status), resources, "", time.Now())
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *Node) payload() (string, []Resource) {
|
||||||
|
if n.cfg.Payload == nil {
|
||||||
|
return "running", nil
|
||||||
|
}
|
||||||
|
return n.cfg.Payload()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *Node) nodeInfo(status string) *NodeInfo {
|
||||||
|
return &NodeInfo{Name: n.cfg.Name, Instance: n.cfg.Instance, Version: n.cfg.Version, Status: status}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *Node) advertiseMsg() Message {
|
||||||
|
status, resources := n.payload()
|
||||||
|
return Message{Proto: Proto, V: Version, Type: TypeAdvertise, Node: n.nodeInfo(status), Resources: resources}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (n *Node) readLoop() {
|
func (n *Node) readLoop() {
|
||||||
@@ -228,32 +246,39 @@ func (n *Node) readLoop() {
|
|||||||
return
|
return
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
log.Printf("[meshdisco] Read error: %v", err)
|
log.Printf("[beacon] Read error: %v", err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
var msg Message
|
msg, ok := Parse(buf[:nRead])
|
||||||
if err := json.Unmarshal(buf[:nRead], &msg); err != nil {
|
if !ok {
|
||||||
// Malformed or foreign traffic on the group — ignore quietly.
|
continue // v1 or foreign traffic on the shared group
|
||||||
continue
|
|
||||||
}
|
|
||||||
if msg.V != ProtocolVersion {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
// Drop our own multicast echo. Loopback is left enabled so that a
|
|
||||||
// single-container test still sees traffic; we filter by identity
|
|
||||||
// instead of relying on the socket option.
|
|
||||||
if msg.Instance != "" && msg.Instance == n.cfg.Instance && msg.Name == n.cfg.Name {
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
|
|
||||||
switch msg.Type {
|
switch msg.Type {
|
||||||
case TypeDiscovery:
|
case TypeProbe:
|
||||||
// Unicast the reply straight back to the prober's source
|
if msg.From == n.cfg.Instance {
|
||||||
// address. The kernel picks the right interface by route, and
|
continue
|
||||||
// the prober's ephemeral source port is where it is listening.
|
}
|
||||||
n.sendAnnounce(src, "running")
|
ad := n.advertiseMsg()
|
||||||
case TypeAnnounce:
|
wanted := len(msg.Want) == 0
|
||||||
|
for _, r := range ad.Resources {
|
||||||
|
if r.MatchesAny(msg.Want) {
|
||||||
|
wanted = true
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if wanted {
|
||||||
|
// Unicast straight back to the prober's source address; its
|
||||||
|
// ephemeral source port is where it is listening.
|
||||||
|
n.send(ad, src)
|
||||||
|
}
|
||||||
|
case TypeAdvertise, TypeBye:
|
||||||
|
// Drop our own multicast echo. Loopback stays enabled so a
|
||||||
|
// single-container test still sees traffic; we filter by identity.
|
||||||
|
if msg.Node.Instance == n.cfg.Instance {
|
||||||
|
continue
|
||||||
|
}
|
||||||
n.record(msg, src)
|
n.record(msg, src)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -269,25 +294,21 @@ func (n *Node) record(msg Message, src net.Addr) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if msg.Status == "stopping" {
|
|
||||||
n.mu.Lock()
|
n.mu.Lock()
|
||||||
delete(n.neighbors, msg.key())
|
defer n.mu.Unlock()
|
||||||
n.mu.Unlock()
|
if msg.Type == TypeBye {
|
||||||
|
delete(n.nodes, msg.Node.Instance)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
n.nodes[msg.Node.Instance] = &seen{node: *msg.Node, resources: msg.Resources, addr: addr, lastSeen: time.Now()}
|
||||||
nb := &Neighbor{Message: msg, Addr: addr, LastSeen: time.Now()}
|
|
||||||
n.mu.Lock()
|
|
||||||
n.neighbors[msg.key()] = nb
|
|
||||||
n.mu.Unlock()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (n *Node) announceLoop() {
|
func (n *Node) advertiseLoop() {
|
||||||
defer n.wg.Done()
|
defer n.wg.Done()
|
||||||
|
|
||||||
// Announce immediately, then probe once so the map fills without waiting
|
// Advertise immediately, then probe once so the map fills without
|
||||||
// a full interval for everyone else's ticker.
|
// waiting a full interval for everyone else's ticker.
|
||||||
n.sendAnnounce(nil, "running")
|
n.broadcast(n.advertiseMsg())
|
||||||
n.Probe()
|
n.Probe()
|
||||||
|
|
||||||
ticker := time.NewTicker(n.cfg.Interval)
|
ticker := time.NewTicker(n.cfg.Interval)
|
||||||
@@ -298,51 +319,32 @@ func (n *Node) announceLoop() {
|
|||||||
case <-n.stopChan:
|
case <-n.stopChan:
|
||||||
return
|
return
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
n.sendAnnounce(nil, "running")
|
n.broadcast(n.advertiseMsg())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (n *Node) buildAnnounce(status string) Message {
|
func encode(msg Message) ([]byte, bool) {
|
||||||
msg := Message{
|
|
||||||
V: ProtocolVersion,
|
|
||||||
Type: TypeAnnounce,
|
|
||||||
Name: n.cfg.Name,
|
|
||||||
Instance: n.cfg.Instance,
|
|
||||||
Role: n.cfg.Role,
|
|
||||||
Version: n.cfg.Version,
|
|
||||||
Status: status,
|
|
||||||
}
|
|
||||||
if n.cfg.Payload != nil {
|
|
||||||
baseURL, st, urls := n.cfg.Payload()
|
|
||||||
msg.BaseUrl = baseURL
|
|
||||||
if st != "" && status == "running" {
|
|
||||||
msg.Status = st
|
|
||||||
}
|
|
||||||
// Only meta-core carries a URLs block; a service announcing one
|
|
||||||
// would let any container on the network impersonate the core.
|
|
||||||
if n.cfg.Role == RoleCore {
|
|
||||||
msg.URLs = urls
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return msg
|
|
||||||
}
|
|
||||||
|
|
||||||
// sendAnnounce multicasts when dst is nil, or unicasts to dst (a probe reply).
|
|
||||||
func (n *Node) sendAnnounce(dst net.Addr, status string) {
|
|
||||||
msg := n.buildAnnounce(status)
|
|
||||||
if dst == nil {
|
|
||||||
n.broadcast(msg)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
body, err := json.Marshal(msg)
|
body, err := json.Marshal(msg)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
return nil, false
|
||||||
|
}
|
||||||
|
if len(body) > MaxDatagram {
|
||||||
|
log.Printf("[beacon] Warning: %d-byte datagram exceeds the fragmentation-safe %d", len(body), MaxDatagram)
|
||||||
|
}
|
||||||
|
return body, true
|
||||||
|
}
|
||||||
|
|
||||||
|
// send unicasts msg to dst (a probe reply).
|
||||||
|
func (n *Node) send(msg Message, dst net.Addr) {
|
||||||
|
body, ok := encode(msg)
|
||||||
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
n.sendMu.Lock()
|
n.sendMu.Lock()
|
||||||
defer n.sendMu.Unlock()
|
defer n.sendMu.Unlock()
|
||||||
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
|
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
|
||||||
log.Printf("[meshdisco] Reply to %s failed: %v", dst, err)
|
log.Printf("[beacon] Reply to %s failed: %v", dst, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -353,8 +355,8 @@ func (n *Node) broadcast(msg Message) {
|
|||||||
if n.pc == nil {
|
if n.pc == nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
body, err := json.Marshal(msg)
|
body, ok := encode(msg)
|
||||||
if err != nil {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port}
|
dst := &net.UDPAddr{IP: n.group, Port: n.cfg.Port}
|
||||||
@@ -363,11 +365,11 @@ func (n *Node) broadcast(msg Message) {
|
|||||||
defer n.sendMu.Unlock()
|
defer n.sendMu.Unlock()
|
||||||
for i := range n.ifaces {
|
for i := range n.ifaces {
|
||||||
if err := n.pc.SetMulticastInterface(&n.ifaces[i]); err != nil {
|
if err := n.pc.SetMulticastInterface(&n.ifaces[i]); err != nil {
|
||||||
log.Printf("[meshdisco] Warning: select interface %s failed: %v", n.ifaces[i].Name, err)
|
log.Printf("[beacon] Warning: select interface %s failed: %v", n.ifaces[i].Name, err)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
|
if _, err := n.pc.WriteTo(body, nil, dst); err != nil {
|
||||||
log.Printf("[meshdisco] Warning: send on %s failed: %v", n.ifaces[i].Name, err)
|
log.Printf("[beacon] Warning: send on %s failed: %v", n.ifaces[i].Name, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -377,7 +379,7 @@ func (n *Node) broadcast(msg Message) {
|
|||||||
func eligibleInterfaces() []net.Interface {
|
func eligibleInterfaces() []net.Interface {
|
||||||
all, err := net.Interfaces()
|
all, err := net.Interfaces()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("[meshdisco] Warning: could not enumerate interfaces: %v", err)
|
log.Printf("[beacon] Warning: could not enumerate interfaces: %v", err)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -402,12 +404,15 @@ func eligibleInterfaces() []net.Interface {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// reusePort sets SO_REUSEADDR so a second listener (a test, or a sidecar) can
|
// reusePort sets SO_REUSEADDR + SO_REUSEPORT so several listeners (a test, a
|
||||||
// bind the same multicast port on the same host.
|
// sidecar, a beacon v1 responder in the same netns) can bind 9099 on one host.
|
||||||
func reusePort(_, _ string, c syscall.RawConn) error {
|
func reusePort(_, _ string, c syscall.RawConn) error {
|
||||||
var serr error
|
var serr error
|
||||||
err := c.Control(func(fd uintptr) {
|
err := c.Control(func(fd uintptr) {
|
||||||
serr = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
|
serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEADDR, 1)
|
||||||
|
if serr == nil {
|
||||||
|
serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
|
||||||
|
}
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
+216
-111
@@ -1,75 +1,62 @@
|
|||||||
// Package meshdisco implements meta-discovery v1: UDP multicast service
|
// Package meshdisco implements beacon v2: local resource advertise / discover
|
||||||
// discovery for the meta-* services.
|
// over UDP multicast (239.255.99.1:9099), the protocol every meta-* service
|
||||||
|
// and SDK-hosted plugin uses to find its neighbours.
|
||||||
//
|
//
|
||||||
// It replaces two file-based mechanisms that both required a shared
|
// A node advertises Resources, each tagged with capability strings
|
||||||
// /meta-core volume:
|
// ("metamesh.core", "metamesh.service/meta-sort", "metamesh.transport/nzb@1");
|
||||||
|
// consumers keep a live view of everyone else's and filter by capability.
|
||||||
|
// meta-core advertises "metamesh.core" carrying its /urls block, which is how
|
||||||
|
// services locate it without a shared volume.
|
||||||
//
|
//
|
||||||
// - /meta-core/locks/kv-leader.info — a one-line file holding meta-core's
|
// The wire protocol is documented in docs/project-architecture/beacon-v2.md.
|
||||||
// API URL, read at boot to locate meta-core.
|
// Keep this package and that doc in step — the Rust (meta-feeder-sdk),
|
||||||
// - /meta-core/services/<name>-<host>.json — per-service registration files
|
// TypeScript and Python implementations are written against the doc.
|
||||||
// aggregated into the dashboard's service nav.
|
|
||||||
//
|
|
||||||
// Both answer the same question — where is meta-core, and who else is on this
|
|
||||||
// network — so one announce packet answers both. Discovery scope becomes the
|
|
||||||
// docker network the container is attached to rather than the volume it
|
|
||||||
// mounts, which is what lets per-app CasaOS stacks and the meta-watch island
|
|
||||||
// see their neighbours without sharing storage.
|
|
||||||
//
|
|
||||||
// The wire protocol is documented in
|
|
||||||
// docs/project-architecture/service-discovery.md. Keep this file and that doc
|
|
||||||
// in step — every other language implementation is written against the doc.
|
|
||||||
package meshdisco
|
package meshdisco
|
||||||
|
|
||||||
import "time"
|
import (
|
||||||
|
"encoding/json"
|
||||||
const (
|
"strings"
|
||||||
// ProtocolVersion is the `v` field on every message. Receivers ignore
|
"time"
|
||||||
// messages carrying a version they don't know.
|
|
||||||
ProtocolVersion = 1
|
|
||||||
|
|
||||||
// DefaultGroup is the IPv4 multicast group. Deliberately NOT beacon's
|
|
||||||
// 239.255.99.1 (sandbox/beacon), so the two protocols cannot cross-talk
|
|
||||||
// if they ever share a network.
|
|
||||||
DefaultGroup = "239.255.77.1"
|
|
||||||
|
|
||||||
// DefaultPort is the UDP port. Deliberately NOT beacon's 9099, for the
|
|
||||||
// same reason.
|
|
||||||
DefaultPort = 9399
|
|
||||||
|
|
||||||
// DefaultInterval is how often an unsolicited announce is emitted.
|
|
||||||
DefaultInterval = 10 * time.Second
|
|
||||||
|
|
||||||
// LivenessFactor multiplies the announce interval to get the staleness
|
|
||||||
// window: a neighbour last seen longer ago than interval*factor is
|
|
||||||
// dropped. This replaces the old heartbeat file + cleaner.go reaper.
|
|
||||||
LivenessFactor = 3
|
|
||||||
|
|
||||||
// ReadBufferSize is the datagram receive buffer. Beacon's reference
|
|
||||||
// implementation uses 1 KiB, which truncates a realistic payload; an
|
|
||||||
// announce carrying the full URLs block is comfortably under 8 KiB.
|
|
||||||
ReadBufferSize = 8192
|
|
||||||
|
|
||||||
// TypeDiscovery is a probe: "who is out there?". Sent at startup and on
|
|
||||||
// a manual refresh, answered immediately by every listener.
|
|
||||||
TypeDiscovery = "discovery"
|
|
||||||
|
|
||||||
// TypeAnnounce is a service describing itself, sent unsolicited on a
|
|
||||||
// ticker and unicast in reply to a probe.
|
|
||||||
TypeAnnounce = "announce"
|
|
||||||
|
|
||||||
// RoleCore marks meta-core itself — the only role that carries a URLs
|
|
||||||
// block. Consumers locating meta-core filter on this.
|
|
||||||
RoleCore = "core"
|
|
||||||
|
|
||||||
// RoleService marks every other meta-* service.
|
|
||||||
RoleService = "service"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// URLs mirrors meta-core's existing GET /urls response (api.URLsResponse)
|
const (
|
||||||
// minus redisUrl, which the api-mediated-access lockdown retired. Embedding it
|
// Proto and Version form the envelope every v2 message carries. Anything
|
||||||
// in the announce makes the announce a drop-in for the /urls hop that
|
// else on the group (beacon v1, junk) is dropped silently.
|
||||||
// LeaderClient performs today; /urls itself stays, since meta-search and
|
Proto = "beacon"
|
||||||
// meta-share still call it.
|
Version = 2
|
||||||
|
|
||||||
|
// DefaultGroup / DefaultPort are beacon's endpoint, on purpose: one
|
||||||
|
// well-known place for every kind of local resource. v1 and v2 coexist
|
||||||
|
// because v2 never sends v1's "discovery" / "announce" types.
|
||||||
|
DefaultGroup = "239.255.99.1"
|
||||||
|
DefaultPort = 9099
|
||||||
|
|
||||||
|
// DefaultInterval is how often an unsolicited advertise is emitted.
|
||||||
|
DefaultInterval = 10 * time.Second
|
||||||
|
|
||||||
|
// LivenessFactor multiplies the interval to get the staleness window.
|
||||||
|
// Expiry is evaluated at read time; there is no reaper.
|
||||||
|
LivenessFactor = 3
|
||||||
|
|
||||||
|
// ReadBufferSize is the receive buffer; MaxDatagram is what a sender
|
||||||
|
// stays under so nothing fragments.
|
||||||
|
ReadBufferSize = 8192
|
||||||
|
MaxDatagram = 1400
|
||||||
|
|
||||||
|
TypeProbe = "probe"
|
||||||
|
TypeAdvertise = "advertise"
|
||||||
|
TypeBye = "bye"
|
||||||
|
|
||||||
|
// CapCore is meta-core's capability; its resource carries Data.urls.
|
||||||
|
CapCore = "metamesh.core"
|
||||||
|
// CapServicePrefix + name is every service's capability.
|
||||||
|
CapServicePrefix = "metamesh.service/"
|
||||||
|
CapAnyService = "metamesh.service/*"
|
||||||
|
)
|
||||||
|
|
||||||
|
// URLs mirrors meta-core's GET /urls response (api.URLsResponse) minus
|
||||||
|
// redisUrl. It travels as the metamesh.core resource's Data.urls, so a service
|
||||||
|
// locates meta-core from the advertise alone.
|
||||||
type URLs struct {
|
type URLs struct {
|
||||||
Hostname string `json:"hostname"`
|
Hostname string `json:"hostname"`
|
||||||
BaseUrl string `json:"baseUrl"`
|
BaseUrl string `json:"baseUrl"`
|
||||||
@@ -78,55 +65,173 @@ type URLs struct {
|
|||||||
WebdavUrlInternal string `json:"webdavUrlInternal"`
|
WebdavUrlInternal string `json:"webdavUrlInternal"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Message is the single wire type: both probe and announce are one JSON
|
// NodeInfo says who is advertising.
|
||||||
// datagram. Everything but `v` and `type` is omitted on a probe.
|
type NodeInfo struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
Instance string `json:"instance"`
|
||||||
|
Version string `json:"version,omitempty"`
|
||||||
|
// Status is "starting" | "running"; absent reads as running.
|
||||||
|
Status string `json:"status,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Resource is one advertised thing.
|
||||||
|
type Resource struct {
|
||||||
|
ID string `json:"id"`
|
||||||
|
Caps []string `json:"caps"`
|
||||||
|
// Endpoints: well-known names http, ui, manifest, mcp. A value starting
|
||||||
|
// with "/" is relative to endpoints.http.
|
||||||
|
Endpoints map[string]string `json:"endpoints,omitempty"`
|
||||||
|
Rev string `json:"rev,omitempty"`
|
||||||
|
// Binds names the one consumer instance this resource belongs to.
|
||||||
|
Binds string `json:"binds,omitempty"`
|
||||||
|
Data json.RawMessage `json:"data,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Matches reports whether any cap matches pattern.
|
||||||
|
func (r Resource) Matches(pattern string) bool {
|
||||||
|
for _, c := range r.Caps {
|
||||||
|
if CapMatches(pattern, c) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
// MatchesAny reports whether any cap matches any pattern; an empty list matches.
|
||||||
|
func (r Resource) MatchesAny(patterns []string) bool {
|
||||||
|
if len(patterns) == 0 {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
for _, p := range patterns {
|
||||||
|
if r.Matches(p) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
// Endpoint returns the named endpoint as an absolute URL.
|
||||||
|
func (r Resource) Endpoint(name string) string {
|
||||||
|
v := r.Endpoints[name]
|
||||||
|
if strings.HasPrefix(v, "/") {
|
||||||
|
base := r.Endpoints["http"]
|
||||||
|
if base == "" {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return strings.TrimRight(base, "/") + v
|
||||||
|
}
|
||||||
|
return v
|
||||||
|
}
|
||||||
|
|
||||||
|
// Message is every v2 datagram.
|
||||||
type Message struct {
|
type Message struct {
|
||||||
|
Proto string `json:"proto"`
|
||||||
V int `json:"v"`
|
V int `json:"v"`
|
||||||
Type string `json:"type"`
|
Type string `json:"type"`
|
||||||
|
Node *NodeInfo `json:"node,omitempty"`
|
||||||
// Name is the service name, e.g. "meta-core", "meta-sort".
|
Resources []Resource `json:"resources,omitempty"`
|
||||||
Name string `json:"name,omitempty"`
|
// From and Want are probe-only.
|
||||||
// Instance distinguishes two instances of the same Name — the container
|
From string `json:"from,omitempty"`
|
||||||
// hostname. Also used to recognise and drop our own multicast echo.
|
Want []string `json:"want,omitempty"`
|
||||||
Instance string `json:"instance,omitempty"`
|
|
||||||
// Role is RoleCore or RoleService.
|
|
||||||
Role string `json:"role,omitempty"`
|
|
||||||
// Version is the service's build version, for display only.
|
|
||||||
Version string `json:"version,omitempty"`
|
|
||||||
// Status is "running" | "starting" | "stopping".
|
|
||||||
Status string `json:"status,omitempty"`
|
|
||||||
|
|
||||||
// BaseUrl is the browser-facing URL the nav menu links to. Built with
|
|
||||||
// the same three-way fallback the file registry used:
|
|
||||||
// META_CORE_PUBLIC_URL -> BASE_URL -> http://<local-ip>:<port>.
|
|
||||||
BaseUrl string `json:"baseUrl,omitempty"`
|
|
||||||
|
|
||||||
// URLs is populated only when Role == RoleCore.
|
|
||||||
URLs *URLs `json:"urls,omitempty"`
|
|
||||||
|
|
||||||
// Token is reserved for a future HMAC over the payload. v1 senders omit
|
|
||||||
// it and v1 receivers ignore it; see the doc's Security section.
|
|
||||||
Token string `json:"token,omitempty"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Neighbor is a Message plus the receive-side facts the sender cannot state
|
// Parse decodes a datagram, returning ok=false for anything that is not a
|
||||||
// about itself.
|
// well-formed beacon v2 message (v1 traffic, other versions, unknown types,
|
||||||
type Neighbor struct {
|
// an advertise/bye without a node).
|
||||||
Message
|
func Parse(b []byte) (Message, bool) {
|
||||||
// Addr is the source address the datagram arrived from. Taken from the
|
var m Message
|
||||||
// packet, never from the payload — a sender does not get to claim its
|
if err := json.Unmarshal(b, &m); err != nil {
|
||||||
// own IP.
|
return m, false
|
||||||
Addr string `json:"addr"`
|
|
||||||
// LastSeen is when we last received an announce from this instance.
|
|
||||||
LastSeen time.Time `json:"lastSeen"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// key identifies a neighbour entry. Two instances of the same service are
|
|
||||||
// distinct neighbours; callers that want one row per service (the nav menu)
|
|
||||||
// collapse by Name themselves.
|
|
||||||
func (m Message) key() string {
|
|
||||||
if m.Instance == "" {
|
|
||||||
return m.Name
|
|
||||||
}
|
}
|
||||||
return m.Name + "|" + m.Instance
|
if m.Proto != Proto || m.V != Version {
|
||||||
|
return m, false
|
||||||
|
}
|
||||||
|
switch m.Type {
|
||||||
|
case TypeProbe:
|
||||||
|
return m, true
|
||||||
|
case TypeAdvertise, TypeBye:
|
||||||
|
if m.Node == nil || m.Node.Name == "" || m.Node.Instance == "" {
|
||||||
|
return m, false
|
||||||
|
}
|
||||||
|
return m, true
|
||||||
|
}
|
||||||
|
return m, false
|
||||||
|
}
|
||||||
|
|
||||||
|
func splitContract(s string) (string, string, bool) {
|
||||||
|
if i := strings.LastIndex(s, "@"); i >= 0 {
|
||||||
|
return s[:i], s[i+1:], true
|
||||||
|
}
|
||||||
|
return s, "", false
|
||||||
|
}
|
||||||
|
|
||||||
|
// CapMatches reports whether capability cap satisfies pattern:
|
||||||
|
// "*" matches everything; an "@N" on the pattern must equal the cap's
|
||||||
|
// contract (none on the pattern ignores it); "x/*" matches any variant of x;
|
||||||
|
// otherwise the bases must be equal.
|
||||||
|
func CapMatches(pattern, cap string) bool {
|
||||||
|
if pattern == "*" {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
pbase, pc, phas := splitContract(pattern)
|
||||||
|
cbase, cc, chas := splitContract(cap)
|
||||||
|
if phas && (!chas || pc != cc) {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
if prefix, ok := strings.CutSuffix(pbase, "/*"); ok {
|
||||||
|
rest, ok := strings.CutPrefix(cbase, prefix)
|
||||||
|
return ok && strings.HasPrefix(rest, "/") && len(rest) > 1
|
||||||
|
}
|
||||||
|
return pbase == cbase
|
||||||
|
}
|
||||||
|
|
||||||
|
// Neighbor is a node heard on the wire, as served by /api/neighbors.
|
||||||
|
type Neighbor struct {
|
||||||
|
NodeInfo
|
||||||
|
Resources []Resource `json:"resources"`
|
||||||
|
// Caps flattens every resource's caps, for consumers that just filter.
|
||||||
|
Caps []string `json:"caps"`
|
||||||
|
// BaseUrl is the metamesh.service/* resource's endpoints.ui — kept so
|
||||||
|
// v1-era readers (the nav menu) keep working.
|
||||||
|
BaseUrl string `json:"baseUrl,omitempty"`
|
||||||
|
// Addr is the packet's source address, never a payload claim.
|
||||||
|
Addr string `json:"addr"`
|
||||||
|
// LastSeen is unix seconds.
|
||||||
|
LastSeen float64 `json:"lastSeen"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Matches reports whether any resource matches pattern.
|
||||||
|
func (n Neighbor) Matches(pattern string) bool {
|
||||||
|
for _, r := range n.Resources {
|
||||||
|
if r.Matches(pattern) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
func newNeighbor(node NodeInfo, resources []Resource, addr string, seen time.Time) Neighbor {
|
||||||
|
nb := Neighbor{
|
||||||
|
NodeInfo: node,
|
||||||
|
Resources: resources,
|
||||||
|
Caps: []string{},
|
||||||
|
Addr: addr,
|
||||||
|
LastSeen: float64(seen.UnixNano()) / 1e9,
|
||||||
|
}
|
||||||
|
if nb.Resources == nil {
|
||||||
|
nb.Resources = []Resource{}
|
||||||
|
}
|
||||||
|
seenCap := map[string]bool{}
|
||||||
|
for _, r := range resources {
|
||||||
|
for _, c := range r.Caps {
|
||||||
|
if !seenCap[c] {
|
||||||
|
seenCap[c] = true
|
||||||
|
nb.Caps = append(nb.Caps, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if nb.BaseUrl == "" && r.Matches(CapAnyService) {
|
||||||
|
nb.BaseUrl = r.Endpoint("ui")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nb
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,140 @@
|
|||||||
|
package meshdisco
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestCapMatches(t *testing.T) {
|
||||||
|
cases := []struct {
|
||||||
|
pattern, cap string
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"metamesh.transport/*", "metamesh.transport/nzb@1", true},
|
||||||
|
{"metamesh.transport/*", "metamesh.transport/ipfs", true},
|
||||||
|
{"metamesh.transport/*", "metamesh.transport", false},
|
||||||
|
{"metamesh.transport/*", "metamesh.transportx/a", false},
|
||||||
|
{"metamesh.transport/*", "metamesh.transport/", false},
|
||||||
|
{"metamesh.transport/nzb", "metamesh.transport/nzb@2", true},
|
||||||
|
{"metamesh.transport/nzb@2", "metamesh.transport/nzb@2", true},
|
||||||
|
{"metamesh.transport/nzb@2", "metamesh.transport/nzb@1", false},
|
||||||
|
{"metamesh.transport/nzb@2", "metamesh.transport/nzb", false},
|
||||||
|
{"metamesh.transport/*@1", "metamesh.transport/nzb@1", true},
|
||||||
|
{"metamesh.core", "metamesh.core", true},
|
||||||
|
{"metamesh.core", "metamesh.core2", false},
|
||||||
|
{"mcp", "mcp/x", false},
|
||||||
|
{"*", "anything@3", true},
|
||||||
|
}
|
||||||
|
for _, c := range cases {
|
||||||
|
if got := CapMatches(c.pattern, c.cap); got != c.want {
|
||||||
|
t.Errorf("CapMatches(%q, %q) = %v, want %v", c.pattern, c.cap, got, c.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestParseRejectsV1AndForeign(t *testing.T) {
|
||||||
|
for _, raw := range []string{
|
||||||
|
`{"type":"discovery"}`,
|
||||||
|
`{"type":"announce","name":"x","tools":[]}`,
|
||||||
|
`{"v":1,"type":"announce","name":"meta-sort"}`,
|
||||||
|
`{"proto":"beacon","v":3,"type":"probe"}`,
|
||||||
|
`{"proto":"beacon","v":2,"type":"announce"}`,
|
||||||
|
`{"proto":"beacon","v":2,"type":"advertise"}`,
|
||||||
|
"\x00garbage",
|
||||||
|
} {
|
||||||
|
if _, ok := Parse([]byte(raw)); ok {
|
||||||
|
t.Errorf("Parse(%q) accepted", raw)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, ok := Parse([]byte(`{"proto":"beacon","v":2,"type":"probe","want":["metamesh.core"]}`)); !ok {
|
||||||
|
t.Error("valid probe rejected")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNeighborRowCarriesCapsAndBaseUrl(t *testing.T) {
|
||||||
|
nb := newNeighbor(
|
||||||
|
NodeInfo{Name: "meta-sort", Instance: "metasort-app"},
|
||||||
|
[]Resource{{ID: "service", Caps: []string{"metamesh.service/meta-sort"}, Endpoints: map[string]string{"ui": "https://x"}}},
|
||||||
|
"10.0.0.1", time.Now(),
|
||||||
|
)
|
||||||
|
if nb.BaseUrl != "https://x" || len(nb.Caps) != 1 || !nb.Matches(CapAnyService) {
|
||||||
|
t.Fatalf("unexpected row: %+v", nb)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// End to end over loopback: a probe with a non-matching want gets no reply;
|
||||||
|
// a matching one gets the advertise; advertise then bye updates the view.
|
||||||
|
func TestProbeAndRecord(t *testing.T) {
|
||||||
|
port := freePort(t)
|
||||||
|
n := New(Config{
|
||||||
|
Name: "meta-core", Instance: "core-1", Port: port, Interval: time.Hour,
|
||||||
|
Payload: func() (string, []Resource) {
|
||||||
|
return "running", []Resource{{ID: "core", Caps: []string{CapCore}}}
|
||||||
|
},
|
||||||
|
})
|
||||||
|
if err := n.Start(); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer n.Stop()
|
||||||
|
|
||||||
|
c, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer c.Close()
|
||||||
|
dst := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: port}
|
||||||
|
buf := make([]byte, ReadBufferSize)
|
||||||
|
|
||||||
|
send := func(m Message) {
|
||||||
|
b, _ := json.Marshal(m)
|
||||||
|
if _, err := c.WriteToUDP(b, dst); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
send(Message{Proto: Proto, V: Version, Type: TypeProbe, From: "cli", Want: []string{"metamesh.service/*"}})
|
||||||
|
c.SetReadDeadline(time.Now().Add(300 * time.Millisecond))
|
||||||
|
if _, _, err := c.ReadFromUDP(buf); err == nil {
|
||||||
|
t.Fatal("non-matching probe got a reply")
|
||||||
|
}
|
||||||
|
|
||||||
|
send(Message{Proto: Proto, V: Version, Type: TypeProbe, From: "cli", Want: []string{CapCore}})
|
||||||
|
c.SetReadDeadline(time.Now().Add(2 * time.Second))
|
||||||
|
nr, _, err := c.ReadFromUDP(buf)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("matching probe got no reply: %v", err)
|
||||||
|
}
|
||||||
|
if m, ok := Parse(buf[:nr]); !ok || m.Type != TypeAdvertise || m.Node.Instance != "core-1" {
|
||||||
|
t.Fatalf("bad reply: %s", buf[:nr])
|
||||||
|
}
|
||||||
|
|
||||||
|
peer := &NodeInfo{Name: "meta-sort", Instance: "sort-1"}
|
||||||
|
send(Message{Proto: Proto, V: Version, Type: TypeAdvertise, Node: peer,
|
||||||
|
Resources: []Resource{{ID: "service", Caps: []string{"metamesh.service/meta-sort"}}}})
|
||||||
|
waitFor(t, func() bool { return len(n.Neighbors()) == 1 })
|
||||||
|
send(Message{Proto: Proto, V: Version, Type: TypeBye, Node: peer})
|
||||||
|
waitFor(t, func() bool { return len(n.Neighbors()) == 0 })
|
||||||
|
}
|
||||||
|
|
||||||
|
func freePort(t *testing.T) int {
|
||||||
|
c, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer c.Close()
|
||||||
|
return c.LocalAddr().(*net.UDPAddr).Port
|
||||||
|
}
|
||||||
|
|
||||||
|
func waitFor(t *testing.T, cond func() bool) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(2 * time.Second)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
if cond() {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
time.Sleep(20 * time.Millisecond)
|
||||||
|
}
|
||||||
|
t.Fatal("condition not reached")
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user