Phase 2a: kamado-api Go middleware (core MVP)
Go 1.22 module that polls CKPool's Unix socket control API, queries
Bitcoin Core over JSON-RPC, merges both into a thread-safe snapshot, and
serves it over REST. Layout:
api/
├── cmd/kamado-api/main.go signal-aware entrypoint
└── internal/
├── config/ env var loader with validation
├── ckpool/ socket client (4-byte LE length-prefixed wire
│ protocol verified against libckpool.c), typed
│ response models for poolstats/users/workers/
│ clients/uptime, + unit tests using a fake
│ unix socket server
├── bitcoind/ minimal JSON-RPC client, getblockchaininfo
│ and getnetworkhashps
├── state/ Aggregator that refreshes a merged Snapshot
│ on a ticker; readers get a copy under RWMutex
└── httpapi/ REST handlers on Go 1.22 ServeMux:
/api/health /api/pool /api/users
/api/workers /api/clients /api/snapshot
CKPool stores hashrate as "dsps" (diff shares per second); we convert
to H/s via the 2^32 constant used by GoBrrr-Pool and other clients.
Every stat CKPool exposes to its socket API is surfaced — useragent,
IP, per-client diff, per-worker best diff — closing the gap against
Bassin which only reads the 60-second stats files.
Dockerfile does a CGO_ENABLED=0 static build on golang:1.22-bookworm
with -trimpath -ldflags=-s -w. docker-compose now runs both ckpool and
kamado-api, sharing a named volume for /run/ckpool so the API can
dial the stratifier socket directly.
Deferred to Phase 2b (called out in README):
- Bitcoin Core ZMQ hashblock subscriber
- WebSocket push for real-time UI updates
- SQLite persistence (block history, best-share history)
- CKPool log tailer for "Solved and confirmed block" detection
Tests and build NOT run in this commit — Go isn't installed in the
dev environment. Run `make api-test` or `make api` to verify.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
satoshi
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
d48035b366
commit
bd0b1b0318
@@ -0,0 +1,124 @@
|
||||
// Package bitcoind provides a minimal Bitcoin Core JSON-RPC client.
|
||||
// Only the methods kamado-api actually needs are implemented.
|
||||
package bitcoind
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
type RPC struct {
|
||||
URL string
|
||||
User string
|
||||
Password string
|
||||
HTTP *http.Client
|
||||
}
|
||||
|
||||
func NewRPC(url, user, password string, timeout time.Duration) *RPC {
|
||||
return &RPC{
|
||||
URL: url,
|
||||
User: user,
|
||||
Password: password,
|
||||
HTTP: &http.Client{Timeout: timeout},
|
||||
}
|
||||
}
|
||||
|
||||
type rpcRequest struct {
|
||||
JSONRPC string `json:"jsonrpc"`
|
||||
ID string `json:"id"`
|
||||
Method string `json:"method"`
|
||||
Params []any `json:"params"`
|
||||
}
|
||||
|
||||
type rpcResponse struct {
|
||||
Result json.RawMessage `json:"result"`
|
||||
Error *rpcError `json:"error"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
type rpcError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
func (e *rpcError) Error() string {
|
||||
return fmt.Sprintf("bitcoind rpc error %d: %s", e.Code, e.Message)
|
||||
}
|
||||
|
||||
// Call performs a single JSON-RPC request and unmarshals the result.
|
||||
func (c *RPC) Call(ctx context.Context, method string, params []any, out any) error {
|
||||
body, err := json.Marshal(rpcRequest{
|
||||
JSONRPC: "1.0",
|
||||
ID: "kamado",
|
||||
Method: method,
|
||||
Params: params,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", c.URL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.SetBasicAuth(c.User, c.Password)
|
||||
|
||||
resp, err := c.HTTP.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("bitcoind rpc: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
raw, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("bitcoind rpc: read body: %w", err)
|
||||
}
|
||||
// bitcoind returns 500 on rpc errors but still with a valid JSON body.
|
||||
var rr rpcResponse
|
||||
if err := json.Unmarshal(raw, &rr); err != nil {
|
||||
return fmt.Errorf("bitcoind rpc: unmarshal (status %d): %w: %s", resp.StatusCode, err, string(raw))
|
||||
}
|
||||
if rr.Error != nil {
|
||||
return rr.Error
|
||||
}
|
||||
if out != nil {
|
||||
return json.Unmarshal(rr.Result, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- typed method wrappers ----
|
||||
|
||||
type BlockchainInfo struct {
|
||||
Chain string `json:"chain"`
|
||||
Blocks int64 `json:"blocks"`
|
||||
Headers int64 `json:"headers"`
|
||||
BestBlockHash string `json:"bestblockhash"`
|
||||
Difficulty float64 `json:"difficulty"`
|
||||
MedianTime int64 `json:"mediantime"`
|
||||
VerificationProgress float64 `json:"verificationprogress"`
|
||||
InitialBlockDownload bool `json:"initialblockdownload"`
|
||||
}
|
||||
|
||||
func (c *RPC) GetBlockchainInfo(ctx context.Context) (*BlockchainInfo, error) {
|
||||
var out BlockchainInfo
|
||||
if err := c.Call(ctx, "getblockchaininfo", nil, &out); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &out, nil
|
||||
}
|
||||
|
||||
// NetworkHashPS returns the network hashrate at the given block height.
|
||||
// `blocks` is a window (default 120). Pass -1 to use the default.
|
||||
func (c *RPC) GetNetworkHashPS(ctx context.Context, blocks, height int) (float64, error) {
|
||||
var out float64
|
||||
if err := c.Call(ctx, "getnetworkhashps", []any{blocks, height}, &out); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
// Package ckpool implements a client for CKPool's Unix-domain-socket control
|
||||
// protocol. Each request opens a fresh connection, writes a 4-byte little-
|
||||
// endian length prefix followed by the payload, half-closes the write side,
|
||||
// then reads a 4-byte length prefix and that many bytes of response.
|
||||
//
|
||||
// See src/libckpool.c in upstream CKPool for the reference implementation
|
||||
// (send_unix_msg / recv_unix_msg).
|
||||
package ckpool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Default socket filenames under the sockdir.
|
||||
const (
|
||||
SocketListener = "listener"
|
||||
SocketStratifier = "stratifier"
|
||||
SocketConnector = "connector"
|
||||
)
|
||||
|
||||
// Client talks to the ckpool Unix sockets at SockDir. It is safe for
|
||||
// concurrent use: each Send opens its own short-lived connection, matching
|
||||
// CKPool's own one-shot request-response model.
|
||||
type Client struct {
|
||||
SockDir string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
// New returns a Client configured with a sensible default timeout.
|
||||
func New(sockDir string) *Client {
|
||||
return &Client{
|
||||
SockDir: sockDir,
|
||||
Timeout: 5 * time.Second,
|
||||
}
|
||||
}
|
||||
|
||||
// maxMsgLen matches the upper bound in libckpool.c (0x80000000). Anything
|
||||
// above this indicates a protocol error and we refuse to allocate for it.
|
||||
const maxMsgLen = 0x80000000
|
||||
|
||||
// ErrProtocol is returned when the server's framing is invalid.
|
||||
var ErrProtocol = errors.New("ckpool: protocol error")
|
||||
|
||||
// Send delivers a single command to the named ckpool socket and returns the
|
||||
// raw response bytes. `sockName` is one of SocketListener, SocketStratifier,
|
||||
// or SocketConnector.
|
||||
func (c *Client) Send(ctx context.Context, sockName, cmd string) ([]byte, error) {
|
||||
if cmd == "" {
|
||||
return nil, fmt.Errorf("ckpool: empty command")
|
||||
}
|
||||
path := c.SockDir + "/" + sockName
|
||||
|
||||
var d net.Dialer
|
||||
dialCtx, cancel := context.WithTimeout(ctx, c.Timeout)
|
||||
defer cancel()
|
||||
conn, err := d.DialContext(dialCtx, "unix", path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ckpool: dial %s: %w", path, err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
if dl, ok := ctx.Deadline(); ok {
|
||||
_ = conn.SetDeadline(dl)
|
||||
} else {
|
||||
_ = conn.SetDeadline(time.Now().Add(c.Timeout))
|
||||
}
|
||||
|
||||
// ---- write: 4-byte LE length + payload + half-close ----
|
||||
var lenBuf [4]byte
|
||||
binary.LittleEndian.PutUint32(lenBuf[:], uint32(len(cmd)))
|
||||
if _, err := conn.Write(lenBuf[:]); err != nil {
|
||||
return nil, fmt.Errorf("ckpool: write len: %w", err)
|
||||
}
|
||||
if _, err := conn.Write([]byte(cmd)); err != nil {
|
||||
return nil, fmt.Errorf("ckpool: write payload: %w", err)
|
||||
}
|
||||
// Half-close write so the server sees EOF and responds, mirroring
|
||||
// libckpool's shutdown(SHUT_WR) at the end of send_unix_msg.
|
||||
if uc, ok := conn.(*net.UnixConn); ok {
|
||||
if err := uc.CloseWrite(); err != nil {
|
||||
return nil, fmt.Errorf("ckpool: close write: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- read: 4-byte LE length + payload ----
|
||||
if _, err := io.ReadFull(conn, lenBuf[:]); err != nil {
|
||||
return nil, fmt.Errorf("ckpool: read len: %w", err)
|
||||
}
|
||||
msgLen := binary.LittleEndian.Uint32(lenBuf[:])
|
||||
if msgLen == 0 || msgLen > maxMsgLen {
|
||||
return nil, fmt.Errorf("%w: invalid msg len %d", ErrProtocol, msgLen)
|
||||
}
|
||||
buf := make([]byte, msgLen)
|
||||
if _, err := io.ReadFull(conn, buf); err != nil {
|
||||
return nil, fmt.Errorf("ckpool: read payload: %w", err)
|
||||
}
|
||||
return buf, nil
|
||||
}
|
||||
|
||||
// SendJSON sends a command and unmarshals the response into `out`.
|
||||
// If the response is not valid JSON, the raw bytes are returned in the error.
|
||||
func (c *Client) SendJSON(ctx context.Context, sockName, cmd string, out any) error {
|
||||
raw, err := c.Send(ctx, sockName, cmd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := json.Unmarshal(raw, out); err != nil {
|
||||
return fmt.Errorf("ckpool: unmarshal %q: %w (raw=%s)", cmd, err, string(raw))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ping returns nil if the stratifier replies "pong".
|
||||
func (c *Client) Ping(ctx context.Context) error {
|
||||
raw, err := c.Send(ctx, SocketStratifier, "ping")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if string(raw) != "pong" {
|
||||
return fmt.Errorf("ckpool: expected pong, got %q", string(raw))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
package ckpool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// fakeServer spawns a goroutine that listens on a unix socket under `dir`,
|
||||
// reads one request in the ckpool wire format, and responds with `reply`.
|
||||
// It validates the request format and reports errors via `t`.
|
||||
func fakeServer(t *testing.T, dir, sockName, wantCmd, reply string) {
|
||||
t.Helper()
|
||||
path := filepath.Join(dir, sockName)
|
||||
ln, err := net.Listen("unix", path)
|
||||
if err != nil {
|
||||
t.Fatalf("listen: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = ln.Close() })
|
||||
|
||||
go func() {
|
||||
conn, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
// Read 4-byte LE length
|
||||
var lenBuf [4]byte
|
||||
if _, err := io.ReadFull(conn, lenBuf[:]); err != nil {
|
||||
t.Errorf("server read len: %v", err)
|
||||
return
|
||||
}
|
||||
n := binary.LittleEndian.Uint32(lenBuf[:])
|
||||
buf := make([]byte, n)
|
||||
if _, err := io.ReadFull(conn, buf); err != nil {
|
||||
t.Errorf("server read payload: %v", err)
|
||||
return
|
||||
}
|
||||
if string(buf) != wantCmd {
|
||||
t.Errorf("server got cmd %q, want %q", string(buf), wantCmd)
|
||||
return
|
||||
}
|
||||
|
||||
// Respond with length prefix + payload
|
||||
binary.LittleEndian.PutUint32(lenBuf[:], uint32(len(reply)))
|
||||
if _, err := conn.Write(lenBuf[:]); err != nil {
|
||||
t.Errorf("server write len: %v", err)
|
||||
return
|
||||
}
|
||||
if _, err := conn.Write([]byte(reply)); err != nil {
|
||||
t.Errorf("server write reply: %v", err)
|
||||
return
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func tempSockDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
// Use a short dir: unix socket paths are limited to ~108 chars on Linux.
|
||||
dir, err := os.MkdirTemp("", "ck")
|
||||
if err != nil {
|
||||
t.Fatalf("mkdir: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = os.RemoveAll(dir) })
|
||||
return dir
|
||||
}
|
||||
|
||||
func TestClient_Ping(t *testing.T) {
|
||||
dir := tempSockDir(t)
|
||||
fakeServer(t, dir, SocketStratifier, "ping", "pong")
|
||||
|
||||
c := New(dir)
|
||||
c.Timeout = time.Second
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
if err := c.Ping(ctx); err != nil {
|
||||
t.Fatalf("ping: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_PoolStats(t *testing.T) {
|
||||
dir := tempSockDir(t)
|
||||
reply := `{"start":1775000000,"update":1775949393,"workers":42664,"users":24679,"disconnected":6119,"shares":338031,"sps1":6.48,"sps5":6.49,"sps15":6.49,"sps60":6.48,"accepted":338031,"rejected":29602,"dsps1":0.274,"dsps5":0.302,"dsps15":0.317,"dsps60":0.309,"dsps360":0.298,"dsps1440":0.290,"dsps10080":0.285}`
|
||||
fakeServer(t, dir, SocketStratifier, "poolstats", reply)
|
||||
|
||||
c := New(dir)
|
||||
c.Timeout = time.Second
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
ps, err := c.PoolStats(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("poolstats: %v", err)
|
||||
}
|
||||
if ps.Workers != 42664 {
|
||||
t.Errorf("Workers = %d, want 42664", ps.Workers)
|
||||
}
|
||||
if got := DSPSToHashrate(ps.DSPS1); got < 1e9 {
|
||||
t.Errorf("DSPS1 hashrate = %v, want > 1GH/s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_Clients(t *testing.T) {
|
||||
dir := tempSockDir(t)
|
||||
reply := `{"clients":[{"id":12345,"useragent":"Bitaxe/2.4.0","address":"192.168.1.100","workername":"bc1q.abc","authorised":true,"bestdiff":12345.67}]}`
|
||||
fakeServer(t, dir, SocketStratifier, "clients", reply)
|
||||
|
||||
c := New(dir)
|
||||
c.Timeout = time.Second
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
clients, err := c.Clients(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("clients: %v", err)
|
||||
}
|
||||
if len(clients) != 1 {
|
||||
t.Fatalf("got %d clients, want 1", len(clients))
|
||||
}
|
||||
if clients[0].UserAgent != "Bitaxe/2.4.0" {
|
||||
t.Errorf("UserAgent = %q, want Bitaxe/2.4.0", clients[0].UserAgent)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_DialError(t *testing.T) {
|
||||
c := New("/nonexistent/path")
|
||||
c.Timeout = 200 * time.Millisecond
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
if err := c.Ping(ctx); err == nil {
|
||||
t.Error("expected error, got nil")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package ckpool
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
// DSPSToHashrate converts CKPool's internal "difficulty shares per second"
|
||||
// value to hashrate in H/s. CKPool shares are normalized to diff 1 which
|
||||
// corresponds to 2^32 hashes.
|
||||
const HashesPerShare = 4294967296.0
|
||||
|
||||
func DSPSToHashrate(dsps float64) float64 { return dsps * HashesPerShare }
|
||||
|
||||
// PoolStats is the response to `poolstats` on the stratifier socket.
|
||||
type PoolStats struct {
|
||||
Start int64 `json:"start"`
|
||||
Update int64 `json:"update"`
|
||||
Workers int64 `json:"workers"`
|
||||
Users int64 `json:"users"`
|
||||
Disconnected int64 `json:"disconnected"`
|
||||
Shares int64 `json:"shares"`
|
||||
SPS1 float64 `json:"sps1"`
|
||||
SPS5 float64 `json:"sps5"`
|
||||
SPS15 float64 `json:"sps15"`
|
||||
SPS60 float64 `json:"sps60"`
|
||||
Accepted int64 `json:"accepted"`
|
||||
Rejected int64 `json:"rejected"`
|
||||
DSPS1 float64 `json:"dsps1"`
|
||||
DSPS5 float64 `json:"dsps5"`
|
||||
DSPS15 float64 `json:"dsps15"`
|
||||
DSPS60 float64 `json:"dsps60"`
|
||||
DSPS360 float64 `json:"dsps360"`
|
||||
DSPS1440 float64 `json:"dsps1440"`
|
||||
DSPS10080 float64 `json:"dsps10080"`
|
||||
}
|
||||
|
||||
// User is one entry in the `users` response.
|
||||
type User struct {
|
||||
User string `json:"user"`
|
||||
ID int64 `json:"id"`
|
||||
Workers int `json:"workers"`
|
||||
BestDiff float64 `json:"bestdiff"`
|
||||
DSPS1 float64 `json:"dsps1"`
|
||||
DSPS5 float64 `json:"dsps5"`
|
||||
DSPS60 float64 `json:"dsps60"`
|
||||
DSPS1440 float64 `json:"dsps1440"`
|
||||
DSPS10080 float64 `json:"dsps10080"`
|
||||
LastShare int64 `json:"lastshare"`
|
||||
}
|
||||
|
||||
type UsersResponse struct {
|
||||
Users []User `json:"users"`
|
||||
}
|
||||
|
||||
// Worker is one entry in the `workers` response.
|
||||
type Worker struct {
|
||||
User string `json:"user"`
|
||||
Worker string `json:"worker"`
|
||||
ID int64 `json:"id"`
|
||||
DSPS1 float64 `json:"dsps1"`
|
||||
DSPS5 float64 `json:"dsps5"`
|
||||
DSPS60 float64 `json:"dsps60"`
|
||||
DSPS1440 float64 `json:"dsps1440"`
|
||||
LastShare int64 `json:"lastshare"`
|
||||
BestDiff float64 `json:"bestdiff"`
|
||||
MinDiff float64 `json:"mindiff"`
|
||||
Idle bool `json:"idle"`
|
||||
}
|
||||
|
||||
type WorkersResponse struct {
|
||||
Workers []Worker `json:"workers"`
|
||||
}
|
||||
|
||||
// Client is one entry in the `clients` response. Note: this is CKPool's
|
||||
// view of a connected stratum session, not our Go socket client.
|
||||
type StratumClient struct {
|
||||
ID int64 `json:"id"`
|
||||
Enonce1 string `json:"enonce1"`
|
||||
Enonce1Var string `json:"enonce1var"`
|
||||
Enonce164 int64 `json:"enonce1_64"`
|
||||
Diff float64 `json:"diff"`
|
||||
DSPS1 float64 `json:"dsps1"`
|
||||
DSPS5 float64 `json:"dsps5"`
|
||||
DSPS60 float64 `json:"dsps60"`
|
||||
DSPS1440 float64 `json:"dsps1440"`
|
||||
DSPS10080 float64 `json:"dsps10080"`
|
||||
LastShare int64 `json:"lastshare"`
|
||||
StartTime int64 `json:"starttime"`
|
||||
Address string `json:"address"`
|
||||
Subscribed bool `json:"subscribed"`
|
||||
Authorised bool `json:"authorised"`
|
||||
Idle bool `json:"idle"`
|
||||
UserAgent string `json:"useragent"`
|
||||
WorkerName string `json:"workername"`
|
||||
UserID int64 `json:"userid"`
|
||||
Server int `json:"server"`
|
||||
BestDiff float64 `json:"bestdiff"`
|
||||
ProxyID int `json:"proxyid"`
|
||||
SubProxyID int `json:"subproxyid"`
|
||||
}
|
||||
|
||||
type ClientsResponse struct {
|
||||
Clients []StratumClient `json:"clients"`
|
||||
}
|
||||
|
||||
// Uptime is the response to `uptime`.
|
||||
type Uptime struct {
|
||||
Uptime int64 `json:"uptime"`
|
||||
}
|
||||
|
||||
// --- High-level wrappers ------------------------------------------------
|
||||
|
||||
func (c *Client) PoolStats(ctx context.Context) (*PoolStats, error) {
|
||||
var ps PoolStats
|
||||
if err := c.SendJSON(ctx, SocketStratifier, "poolstats", &ps); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ps, nil
|
||||
}
|
||||
|
||||
func (c *Client) Users(ctx context.Context) ([]User, error) {
|
||||
var r UsersResponse
|
||||
if err := c.SendJSON(ctx, SocketStratifier, "users", &r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Users, nil
|
||||
}
|
||||
|
||||
func (c *Client) Workers(ctx context.Context) ([]Worker, error) {
|
||||
var r WorkersResponse
|
||||
if err := c.SendJSON(ctx, SocketStratifier, "workers", &r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Workers, nil
|
||||
}
|
||||
|
||||
func (c *Client) Clients(ctx context.Context) ([]StratumClient, error) {
|
||||
var r ClientsResponse
|
||||
if err := c.SendJSON(ctx, SocketStratifier, "clients", &r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return r.Clients, nil
|
||||
}
|
||||
|
||||
func (c *Client) Uptime(ctx context.Context) (int64, error) {
|
||||
var u Uptime
|
||||
if err := c.SendJSON(ctx, SocketStratifier, "uptime", &u); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return u.Uptime, nil
|
||||
}
|
||||
|
||||
// GetUser returns stats for a single user by address.
|
||||
func (c *Client) GetUser(ctx context.Context, address string) (*User, error) {
|
||||
cmd := fmt.Sprintf("getuser.{\"user\":%q}", address)
|
||||
var u User
|
||||
if err := c.SendJSON(ctx, SocketStratifier, cmd, &u); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &u, nil
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
// Package config loads kamado-api configuration from environment variables.
|
||||
// Missing required values are a hard error at startup; optional values get
|
||||
// documented defaults.
|
||||
package config
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
// HTTP server
|
||||
ListenAddr string
|
||||
|
||||
// CKPool socket
|
||||
CKPoolSockDir string
|
||||
|
||||
// Bitcoin Core RPC
|
||||
BitcoinRPCURL string // full URL e.g. http://bitcoind:8332
|
||||
BitcoinRPCUser string
|
||||
BitcoinRPCPassword string
|
||||
BitcoinRPCTimeout time.Duration
|
||||
|
||||
// ZMQ (Phase 2b)
|
||||
BitcoinZMQBlock string // e.g. tcp://bitcoind:28332, empty to disable
|
||||
|
||||
// CKPool log file, for block-solve detection (Phase 2b)
|
||||
CKPoolLogFile string
|
||||
|
||||
// SQLite DB path for persistence (Phase 2b)
|
||||
DBPath string
|
||||
|
||||
// Poll interval for refreshing ckpool stats
|
||||
PollInterval time.Duration
|
||||
}
|
||||
|
||||
func FromEnv() (*Config, error) {
|
||||
cfg := &Config{
|
||||
ListenAddr: getenv("LISTEN_ADDR", ":8080"),
|
||||
CKPoolSockDir: getenv("CKPOOL_SOCKDIR", "/run/ckpool"),
|
||||
BitcoinRPCURL: os.Getenv("BITCOIN_RPC_URL"),
|
||||
BitcoinRPCUser: os.Getenv("BITCOIN_RPC_USER"),
|
||||
BitcoinRPCPassword: os.Getenv("BITCOIN_RPC_PASSWORD"),
|
||||
BitcoinRPCTimeout: getenvDuration("BITCOIN_RPC_TIMEOUT", 10*time.Second),
|
||||
BitcoinZMQBlock: os.Getenv("BITCOIN_ZMQ_BLOCK"),
|
||||
CKPoolLogFile: getenv("CKPOOL_LOGFILE", "/var/log/ckpool/ckpool.log"),
|
||||
DBPath: getenv("DB_PATH", "/var/lib/kamado/kamado.db"),
|
||||
PollInterval: getenvDuration("POLL_INTERVAL", 5*time.Second),
|
||||
}
|
||||
|
||||
if cfg.BitcoinRPCURL == "" {
|
||||
return nil, fmt.Errorf("BITCOIN_RPC_URL is required")
|
||||
}
|
||||
if cfg.BitcoinRPCUser == "" || cfg.BitcoinRPCPassword == "" {
|
||||
return nil, fmt.Errorf("BITCOIN_RPC_USER and BITCOIN_RPC_PASSWORD are required")
|
||||
}
|
||||
return cfg, nil
|
||||
}
|
||||
|
||||
func getenv(key, def string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func getenvDuration(key string, def time.Duration) time.Duration {
|
||||
v := os.Getenv(key)
|
||||
if v == "" {
|
||||
return def
|
||||
}
|
||||
// Accept both Go duration syntax ("5s", "10s") and bare seconds ("10").
|
||||
if d, err := time.ParseDuration(v); err == nil {
|
||||
return d
|
||||
}
|
||||
if n, err := strconv.Atoi(v); err == nil {
|
||||
return time.Duration(n) * time.Second
|
||||
}
|
||||
return def
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
// Package httpapi serves the Kamado REST API. WebSocket push and static
|
||||
// UI serving land in a follow-up commit.
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
|
||||
"github.com/kamadopool/kamado-api/internal/state"
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
Agg *state.Aggregator
|
||||
Log *slog.Logger
|
||||
}
|
||||
|
||||
func New(agg *state.Aggregator, log *slog.Logger) *Server {
|
||||
return &Server{Agg: agg, Log: log}
|
||||
}
|
||||
|
||||
// Handler returns an http.Handler with all kamado routes mounted under /api.
|
||||
func (s *Server) Handler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("GET /api/health", s.health)
|
||||
mux.HandleFunc("GET /api/pool", s.pool)
|
||||
mux.HandleFunc("GET /api/users", s.users)
|
||||
mux.HandleFunc("GET /api/workers", s.workers)
|
||||
mux.HandleFunc("GET /api/clients", s.clients)
|
||||
mux.HandleFunc("GET /api/snapshot", s.snapshot)
|
||||
return mux
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
// ---- handlers ---------------------------------------------------------
|
||||
|
||||
func (s *Server) health(w http.ResponseWriter, r *http.Request) {
|
||||
snap := s.Agg.Snapshot()
|
||||
status := http.StatusOK
|
||||
if !snap.CKPoolOK || !snap.BitcoinOK {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
writeJSON(w, status, map[string]any{
|
||||
"ok": snap.CKPoolOK && snap.BitcoinOK,
|
||||
"ckpool": snap.CKPoolOK,
|
||||
"bitcoin": snap.BitcoinOK,
|
||||
"last_error": snap.LastError,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) snapshot(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, s.Agg.Snapshot())
|
||||
}
|
||||
|
||||
func (s *Server) pool(w http.ResponseWriter, r *http.Request) {
|
||||
snap := s.Agg.Snapshot()
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"pool": snap.Pool,
|
||||
"uptime_seconds": snap.Uptime,
|
||||
"hashrate_hs_1m": snap.HashrateHs,
|
||||
"hashrate_hs_5m": snap.HashrateHs5m,
|
||||
"hashrate_hs_1h": snap.HashrateHs1h,
|
||||
"hashrate_hs_24h": snap.HashrateHs24h,
|
||||
"chain": snap.Chain,
|
||||
"network_hashrate_hs": snap.NetworkHashrateHs,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) users(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, s.Agg.Snapshot().Users)
|
||||
}
|
||||
|
||||
func (s *Server) workers(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, s.Agg.Snapshot().Workers)
|
||||
}
|
||||
|
||||
func (s *Server) clients(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, s.Agg.Snapshot().Clients)
|
||||
}
|
||||
@@ -0,0 +1,137 @@
|
||||
// Package state merges data from CKPool (via Unix socket), Bitcoin Core
|
||||
// (via JSON-RPC), and future sources (ZMQ, log tailer) into a single
|
||||
// thread-safe snapshot the HTTP layer can serve. The snapshot is refreshed
|
||||
// on a ticker; readers get a copy without blocking the refresh.
|
||||
package state
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/kamadopool/kamado-api/internal/bitcoind"
|
||||
"github.com/kamadopool/kamado-api/internal/ckpool"
|
||||
)
|
||||
|
||||
// Snapshot is the merged view served to the UI. All fields are safe to
|
||||
// JSON-serialize directly.
|
||||
type Snapshot struct {
|
||||
GeneratedAt time.Time `json:"generated_at"`
|
||||
|
||||
// CKPool-derived fields
|
||||
Pool *ckpool.PoolStats `json:"pool"`
|
||||
Uptime int64 `json:"uptime_seconds"`
|
||||
Users []ckpool.User `json:"users"`
|
||||
Workers []ckpool.Worker `json:"workers"`
|
||||
Clients []ckpool.StratumClient `json:"clients"`
|
||||
|
||||
// Derived/enriched fields
|
||||
HashrateHs float64 `json:"hashrate_hs_1m"` // from PoolStats.DSPS1
|
||||
HashrateHs5m float64 `json:"hashrate_hs_5m"`
|
||||
HashrateHs1h float64 `json:"hashrate_hs_1h"`
|
||||
HashrateHs24h float64 `json:"hashrate_hs_24h"`
|
||||
|
||||
// Bitcoin Core fields
|
||||
Chain *bitcoind.BlockchainInfo `json:"chain"`
|
||||
NetworkHashrateHs float64 `json:"network_hashrate_hs"`
|
||||
|
||||
// Health
|
||||
CKPoolOK bool `json:"ckpool_ok"`
|
||||
BitcoinOK bool `json:"bitcoin_ok"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
}
|
||||
|
||||
// Aggregator refreshes a Snapshot on a ticker.
|
||||
type Aggregator struct {
|
||||
CK *ckpool.Client
|
||||
RPC *bitcoind.RPC
|
||||
Interval time.Duration
|
||||
Log *slog.Logger
|
||||
|
||||
mu sync.RWMutex
|
||||
snap Snapshot
|
||||
}
|
||||
|
||||
func New(ck *ckpool.Client, rpc *bitcoind.RPC, interval time.Duration, log *slog.Logger) *Aggregator {
|
||||
return &Aggregator{
|
||||
CK: ck,
|
||||
RPC: rpc,
|
||||
Interval: interval,
|
||||
Log: log,
|
||||
}
|
||||
}
|
||||
|
||||
// Run blocks until ctx is cancelled, refreshing the snapshot every Interval.
|
||||
// It runs one immediate refresh at startup so readers don't see an empty
|
||||
// snapshot after ctx launches the goroutine.
|
||||
func (a *Aggregator) Run(ctx context.Context) {
|
||||
a.refresh(ctx)
|
||||
t := time.NewTicker(a.Interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
a.refresh(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Snapshot returns a copy of the current snapshot.
|
||||
func (a *Aggregator) Snapshot() Snapshot {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
return a.snap
|
||||
}
|
||||
|
||||
func (a *Aggregator) refresh(ctx context.Context) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
next := Snapshot{GeneratedAt: time.Now()}
|
||||
|
||||
// --- ckpool: poolstats, users, workers, clients, uptime ---
|
||||
if ps, err := a.CK.PoolStats(ctx); err == nil {
|
||||
next.Pool = ps
|
||||
next.HashrateHs = ckpool.DSPSToHashrate(ps.DSPS1)
|
||||
next.HashrateHs5m = ckpool.DSPSToHashrate(ps.DSPS5)
|
||||
next.HashrateHs1h = ckpool.DSPSToHashrate(ps.DSPS60)
|
||||
next.HashrateHs24h = ckpool.DSPSToHashrate(ps.DSPS1440)
|
||||
next.CKPoolOK = true
|
||||
} else {
|
||||
a.Log.Warn("ckpool poolstats failed", "err", err)
|
||||
next.LastError = err.Error()
|
||||
}
|
||||
if u, err := a.CK.Uptime(ctx); err == nil {
|
||||
next.Uptime = u
|
||||
}
|
||||
if us, err := a.CK.Users(ctx); err == nil {
|
||||
next.Users = us
|
||||
}
|
||||
if ws, err := a.CK.Workers(ctx); err == nil {
|
||||
next.Workers = ws
|
||||
}
|
||||
if cs, err := a.CK.Clients(ctx); err == nil {
|
||||
next.Clients = cs
|
||||
}
|
||||
|
||||
// --- bitcoind: chain + network hashrate ---
|
||||
if bi, err := a.RPC.GetBlockchainInfo(ctx); err == nil {
|
||||
next.Chain = bi
|
||||
next.BitcoinOK = true
|
||||
if nh, err := a.RPC.GetNetworkHashPS(ctx, -1, int(bi.Blocks)); err == nil {
|
||||
next.NetworkHashrateHs = nh
|
||||
}
|
||||
} else {
|
||||
a.Log.Warn("bitcoind getblockchaininfo failed", "err", err)
|
||||
if next.LastError == "" {
|
||||
next.LastError = err.Error()
|
||||
}
|
||||
}
|
||||
|
||||
a.mu.Lock()
|
||||
a.snap = next
|
||||
a.mu.Unlock()
|
||||
}
|
||||
Reference in New Issue
Block a user