mcp io

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## v1.7.2
+
+### Features
+- **Two MCP tools for I/O**, `io_summary` and `io_trend`: an assistant can ask who is doing the reading and writing — client load, autovacuum or the checkpointer — and when it started, with a knowledge-base page on how to read the counters. An empty answer names its cause and a period broken by a statistics reset carries the part that was measured. Write-ahead log I/O times follow `track_wal_io_timing`, so they are read even where `track_io_timing` is off; needs PostgreSQL 16 or newer.
+
 ## v1.7.1
 
 ### Features

CHANGELOG.ru.md

@@ -1,5 +1,10 @@
 # История изменений
 
+## v1.7.2
+
+### Фичи
+- **Два MCP-инструмента по вводу-выводу** — `io_summary` и `io_trend`: ассистент может спросить, кто читает и пишет — клиентская нагрузка, автовакуум или чекпойнтер — и когда это началось, плюс страница базы знаний о том, как читать счётчики. Пустой ответ называет причину, а период, разорванный сбросом статистики, несёт измеренную часть. Времена ввода-вывода журнала предзаписи берутся по `track_wal_io_timing` и читаются даже при выключенном `track_io_timing`; нужен PostgreSQL 16 или новее.
+
 ## v1.7.1
 
 ### Фичи

backend/internal/http/v1_io.go

@@ -181,18 +181,42 @@ func ioHistoryMeta(instance string, earliest, latest *time.Time, metas []statio.
 		}
 	}
 
+	setWALTiming(&meta, metas)
+
 	return meta
 }
 
+// Captures older than the track_wal_io_timing column carry no value at all;
+// they are skipped, so an unknown state never counts as a toggle.
+func setWALTiming(meta *serverhttp.IOHistoryMeta, metas []statio.Meta) {
+	var first *bool
+
+	for _, m := range metas {
+		if m.TrackWALIOTiming == nil {
+			continue
+		}
+
+		switch {
+		case first == nil:
+			first = m.TrackWALIOTiming
+		case *m.TrackWALIOTiming != *first:
+			meta.TrackWalIoTimingChanged = true
+		}
+
+		meta.TrackWalIoTiming = *m.TrackWALIOTiming
+	}
+}
+
 func ioSnapshotToAPI(instance string, snap statio.Snapshot) serverhttp.IOSnapshot {
 	out := serverhttp.IOSnapshot{
-		Instance:      instance,
-		CapturedAt:    snap.CapturedAt,
-		VersionNum:    snap.VersionNum,
-		OpBytes:       snap.OpBytes,
-		TrackIoTiming: snap.TrackIOTiming,
-		StatsReset:    snap.StatsReset,
-		Rows:          make([]serverhttp.IORow, 0, len(snap.Rows)),
+		Instance:         instance,
+		CapturedAt:       snap.CapturedAt,
+		VersionNum:       snap.VersionNum,
+		OpBytes:          snap.OpBytes,
+		TrackIoTiming:    snap.TrackIOTiming,
+		TrackWalIoTiming: snap.TrackWALIOTiming,
+		StatsReset:       snap.StatsReset,
+		Rows:             make([]serverhttp.IORow, 0, len(snap.Rows)),
 	}
 
 	for _, r := range snap.Rows {

backend/internal/http/v1_io_test.go

@@ -147,6 +147,57 @@ func TestIOHistoryMetaReportsEpochShifts(t *testing.T) {
 	}
 }
 
+// The two timing GUCs are reported apart: WAL times survive track_io_timing off.
+func TestIOHistoryMetaSeparatesWALTiming(t *testing.T) {
+	t.Parallel()
+
+	at := time.Date(2026, 8, 22, 3, 0, 0, 0, time.UTC)
+	earliest, latest := at.Add(-24*time.Hour), at
+
+	on := true
+	metas := []statio.Meta{
+		{CapturedAt: at.Add(-2 * time.Minute), VersionNum: 180001, TrackWALIOTiming: &on},
+		{CapturedAt: at.Add(-time.Minute), VersionNum: 180001, TrackWALIOTiming: &on},
+	}
+
+	meta := ioHistoryMeta("h1", &earliest, &latest, metas)
+
+	if meta.TrackIoTiming || meta.TrackIoTimingChanged {
+		t.Error("relation timing was off the whole period")
+	}
+
+	if !meta.TrackWalIoTiming {
+		t.Error("track_wal_io_timing must reflect the newest capture")
+	}
+
+	if meta.TrackWalIoTimingChanged {
+		t.Error("track_wal_io_timing held steady")
+	}
+}
+
+func TestIOHistoryMetaIgnoresUnrecordedWALTiming(t *testing.T) {
+	t.Parallel()
+
+	at := time.Date(2026, 8, 22, 3, 0, 0, 0, time.UTC)
+	earliest, latest := at.Add(-24*time.Hour), at
+
+	on := true
+	metas := []statio.Meta{
+		{CapturedAt: at.Add(-2 * time.Minute), VersionNum: 180001},
+		{CapturedAt: at.Add(-time.Minute), VersionNum: 180001, TrackWALIOTiming: &on},
+	}
+
+	meta := ioHistoryMeta("h1", &earliest, &latest, metas)
+
+	if meta.TrackWalIoTimingChanged {
+		t.Error("a capture that recorded no track_wal_io_timing is not a toggle")
+	}
+
+	if !meta.TrackWalIoTiming {
+		t.Error("the newest recorded value must be reported")
+	}
+}
+
 func TestIOHistoryMetaWithoutSnapshots(t *testing.T) {
 	t.Parallel()
 

backend/internal/mcpserver/client.go

@@ -509,6 +509,45 @@ func (d *DashaClient) WaitEvents(ctx context.Context, cluster, instance string)
 	return pick(r.JSON200, r.HTTPResponse, "wait_events")
 }
 
+// IOHistory returns the pg_stat_io deltas of one host over a period.
+func (d *DashaClient) IOHistory(ctx context.Context, params *apiclient.GetIOHistoryParams) (*apiclient.IOHistory, error) {
+	r, err := d.api.GetIOHistoryWithResponse(ctx, params, d.editor(ctx))
+	if err != nil {
+		return nil, wrapErr("io_history", err)
+	}
+
+	if r.JSON200 == nil && r.HTTPResponse != nil && r.HTTPResponse.StatusCode == http.StatusNotImplemented {
+		return nil, errors.New("dasha: I/O history needs snapshot storage, which this Dasha has not " +
+			"configured (501) — pg_stat_io history is unavailable on this deployment")
+	}
+
+	if r.JSON200 == nil {
+		return nil, statusError("io_history", r.HTTPResponse)
+	}
+
+	return r.JSON200, nil
+}
+
+// IOSupported probes the live route, which answers 501 below PostgreSQL 16.
+func (d *DashaClient) IOSupported(ctx context.Context, cluster, instance string) (bool, error) {
+	r, err := d.api.GetIOCurrentWithResponse(ctx, &apiclient.GetIOCurrentParams{
+		ClusterName: cluster, Instance: instance,
+	}, d.editor(ctx))
+	if err != nil {
+		return false, wrapErr("io_current", err)
+	}
+
+	if r.HTTPResponse != nil && r.HTTPResponse.StatusCode == http.StatusNotImplemented {
+		return false, nil
+	}
+
+	if r.JSON200 == nil {
+		return false, statusError("io_current", r.HTTPResponse)
+	}
+
+	return true, nil
+}
+
 // SchemaLint returns the structural defects of one database's schema: code,
 // level, object and the numbers behind each finding. Reads the system catalog
 // only. Optional level filter ("error", "warning", "notice") narrows the list

backend/internal/mcpserver/e2e_test.go

@@ -5,6 +5,8 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/modelcontextprotocol/go-sdk/mcp"
@@ -240,3 +242,160 @@ func firstText(res *mcp.CallToolResult) string {
 
 	return ""
 }
+
+// TestE2E_IOTools drives both pg_stat_io tools over a real client session: the
+// tools are advertised, arguments reach the history endpoint, and the shaped
+// result comes back as readable JSON.
+func TestE2E_IOTools(t *testing.T) {
+	t.Parallel()
+
+	var (
+		mu        sync.Mutex
+		lastQuery string
+	)
+
+	query := func() string {
+		mu.Lock()
+		defer mu.Unlock()
+
+		return lastQuery
+	}
+
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api/io/history" {
+			w.WriteHeader(http.StatusNotFound)
+
+			return
+		}
+
+		mu.Lock()
+		lastQuery = r.URL.RawQuery
+		mu.Unlock()
+
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(ioHistoryJSON))
+	}))
+	defer backend.Close()
+
+	client, err := NewDashaClient(Config{DashaURL: backend.URL, Token: "t"}) //nolint:exhaustruct
+	if err != nil {
+		t.Fatalf("NewDashaClient: %v", err)
+	}
+
+	ctx := context.Background()
+	st, ct := mcp.NewInMemoryTransports()
+
+	ss, err := NewMCPServer(client, "test", "en").Connect(ctx, st, nil)
+	if err != nil {
+		t.Fatalf("server connect: %v", err)
+	}
+	defer ss.Close()
+
+	c := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "v0"}, nil) //nolint:exhaustruct
+
+	cs, err := c.Connect(ctx, ct, nil)
+	if err != nil {
+		t.Fatalf("client connect: %v", err)
+	}
+	defer cs.Close()
+
+	lt, err := cs.ListTools(ctx, nil)
+	if err != nil {
+		t.Fatalf("ListTools: %v", err)
+	}
+
+	for _, want := range []string{"io_summary", "io_trend"} {
+		if !hasTool(lt.Tools, want) {
+			t.Errorf("tool %q not advertised", want)
+		}
+	}
+
+	res, err := cs.CallTool(ctx, &mcp.CallToolParams{ //nolint:exhaustruct
+		Name:      "io_summary",
+		Arguments: map[string]any{"cluster": "demo", "instance": "h1", "group_by": "full", "top": 5},
+	})
+	if err != nil {
+		t.Fatalf("CallTool(io_summary): %v", err)
+	}
+
+	if res.IsError {
+		t.Fatalf("io_summary returned IsError: %s", firstText(res))
+	}
+
+	if got := firstText(res); !strings.Contains(got, `"ranked_by"`) || !strings.Contains(got, "vacuum") {
+		t.Errorf("result = %q, want a ranked table naming the vacuum context", got)
+	}
+
+	if q := query(); !strings.Contains(q, "group_by=full") || !strings.Contains(q, "points=1") {
+		t.Errorf("history query = %q, want group_by=full and points=1", q)
+	}
+
+	res, err = cs.CallTool(ctx, &mcp.CallToolParams{ //nolint:exhaustruct
+		Name:      "io_trend",
+		Arguments: map[string]any{"cluster": "demo", "instance": "h1", "since": "6h"},
+	})
+	if err != nil {
+		t.Fatalf("CallTool(io_trend): %v", err)
+	}
+
+	if res.IsError {
+		t.Fatalf("io_trend returned IsError: %s", firstText(res))
+	}
+
+	if q := query(); !strings.Contains(q, "points=24") || !strings.Contains(q, "group_by=context") {
+		t.Errorf("history query = %q, want the trend defaults", q)
+	}
+}
+
+// TestE2E_IOToolsRejectBadArgs confirms local validation answers as a readable
+// isError result without ever reaching Dasha.
+func TestE2E_IOToolsRejectBadArgs(t *testing.T) {
+	t.Parallel()
+
+	var called atomic.Bool
+
+	backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		called.Store(true)
+
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer backend.Close()
+
+	client, err := NewDashaClient(Config{DashaURL: backend.URL, Token: "t"}) //nolint:exhaustruct
+	if err != nil {
+		t.Fatalf("NewDashaClient: %v", err)
+	}
+
+	ctx := context.Background()
+	st, ct := mcp.NewInMemoryTransports()
+
+	ss, err := NewMCPServer(client, "test", "en").Connect(ctx, st, nil)
+	if err != nil {
+		t.Fatalf("server connect: %v", err)
+	}
+	defer ss.Close()
+
+	c := mcp.NewClient(&mcp.Implementation{Name: "test-client", Version: "v0"}, nil) //nolint:exhaustruct
+
+	cs, err := c.Connect(ctx, ct, nil)
+	if err != nil {
+		t.Fatalf("client connect: %v", err)
+	}
+	defer cs.Close()
+
+	res, err := cs.CallTool(ctx, &mcp.CallToolParams{ //nolint:exhaustruct
+		Name:      "io_summary",
+		Arguments: map[string]any{"cluster": "demo", "instance": "h1", "group_by": "object"},
+	})
+	if err != nil {
+		t.Fatalf("CallTool: %v", err)
+	}
+
+	if !res.IsError {
+		t.Errorf("an unknown group_by must be refused")
+	}
+
+	if called.Load() {
+		t.Errorf("invalid arguments must not reach Dasha")
+	}
+}