diff --git a/api.go b/api.go index 47558678d..628feb16d 100644 --- a/api.go +++ b/api.go @@ -23,6 +23,7 @@ import ( "github.com/molecula/featurebase/v2/disco" "github.com/molecula/featurebase/v2/ingest" + "github.com/molecula/featurebase/v2/rbf" //"github.com/molecula/featurebase/v2/pg" "github.com/molecula/featurebase/v2/pql" @@ -3156,6 +3157,21 @@ func (api *API) Plan(ctx context.Context, q string) (*Stmt, error) { return api.server.PlanSQL(ctx, q) } +func (api *API) RBFDebugInfo() map[string]*rbf.DebugInfo { + infos := make(map[string]*rbf.DebugInfo) + + for key, dbShard := range api.holder.Txf().dbPerShard.Flatmap { + wrapper, ok := dbShard.W.(*RbfDBWrapper) + if !ok { + continue + } + + skey := fmt.Sprintf("%s/%d", key.index, key.shard) + infos[skey] = wrapper.db.DebugInfo() + } + return infos +} + type serverInfo struct { ShardWidth uint64 `json:"shardWidth"` ReplicaN int `json:"replicaN"` diff --git a/api_test.go b/api_test.go index 9ffdea841..d42aca667 100644 --- a/api_test.go +++ b/api_test.go @@ -1415,3 +1415,25 @@ func TestVariousApiTranslateCalls(t *testing.T) { */ } } + +func TestAPI_RBFDebugInfo(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := test.MustRunCluster(t, 1, + []server.CommandOption{ + server.OptCommandServerOptions( + pilosa.OptServerNodeID("node0"), + pilosa.OptServerClusterHasher(&offsetModHasher{}), + pilosa.OptServerOpenTranslateReader(http.GetOpenTranslateReaderFunc(nil)), + )}, + ) + defer c.Close() + + coord := c.GetPrimary() + + if _, err := coord.API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil { + t.Fatal(err) + } else if infos := coord.API.RBFDebugInfo(); infos == nil { + t.Fatal("expected info") + } +} diff --git a/http/handler.go b/http/handler.go index 958f29662..05b65aa58 100644 --- a/http/handler.go +++ b/http/handler.go @@ -441,6 +441,9 @@ func newRouter(handler *Handler) http.Handler { router.HandleFunc("/internal/idalloc/data", handler.handleIDAllocData).Methods("GET").Name("IDAllocData") router.HandleFunc("/internal/restore/{index}/{shardID}", handler.handlePostRestore).Methods("POST").Name("Restore") + + router.HandleFunc("/internal/debug/rbf", handler.handleGetInternalDebugRBFJSON).Methods("GET").Name("GetInternalDebugRBFJSON") + // endpoints for collecting cpu profiles from a chosen begin point to // when the client wants to stop. Used for profiling imports that // could be long or short. @@ -2064,6 +2067,18 @@ func validateProtobufHeader(r *http.Request) (error string, code int) { return } +// handleGetInternalDebugRBFJSON handles /internal/debug/rbf requests. +func (h *Handler) handleGetInternalDebugRBFJSON(w http.ResponseWriter, r *http.Request) { + buf, err := json.MarshalIndent(h.api.RBFDebugInfo(), "", " ") + if err != nil { + http.Error(w, "marshal json: "+err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + w.Write(buf) +} + // handleGetMetricsJSON handles /metrics.json requests, translating text metrics results to more consumable JSON. func (h *Handler) handleGetMetricsJSON(w http.ResponseWriter, r *http.Request) { if !validHeaderAcceptJSON(r.Header) { diff --git a/rbf/db.go b/rbf/db.go index 65dd4fbea..6e4955f94 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -7,6 +7,8 @@ import ( "io" "os" "path/filepath" + "runtime/debug" + "sort" "sync" "syscall" @@ -637,6 +639,7 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { pageMap: db.pageMap, walPageN: db.walPageN, writable: writable, + stack: debug.Stack(), // DEBUG DeleteEmptyContainer: true, } @@ -815,6 +818,20 @@ func (db *DB) getCursor(tx *Tx) *Cursor { return c } +func (db *DB) DebugInfo() *DebugInfo { + info := &DebugInfo{Path: db.Path} + for tx := range db.txs { + info.Txs = append(info.Txs, tx.DebugInfo()) + } + sort.Slice(info.Txs, func(i, j int) bool { return info.Txs[i].Ptr < info.Txs[j].Ptr }) + return info +} + +type DebugInfo struct { + Path string `json:"path"` + Txs []*TxDebugInfo `json:"txs"` +} + // Shared pool for in-memory database pages. // These are used before being flushed to disk. var pagePool = &sync.Pool{ diff --git a/rbf/db_test.go b/rbf/db_test.go index c0eb3b3c7..8bae550bb 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -339,6 +339,21 @@ func TestDB_MultiTx(t *testing.T) { } } +func TestDB_DebugInfo(t *testing.T) { + db := MustOpenDB(t) + defer MustCloseDB(t, db) + + tx := MustBegin(t, db, true) + defer tx.Rollback() + + info := db.DebugInfo() + if got, want := info.Path, db.Path; got != want { + t.Fatalf("Path=%q, want %q", got, want) + } else if got, want := len(info.Txs), 1; got != want { + t.Fatalf("len(Txs)=%d, want %d", got, want) + } +} + // premake pool of random values const randPool = (1 << 18) diff --git a/rbf/tx.go b/rbf/tx.go index 5c6c7f7c6..5d731fe4f 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -65,6 +65,9 @@ type Tx struct { // manages to trigger a *deallocation* (which I don't think should be // happening), we'll process that one after the current list is processed. pendingFreelistAdds []uint32 + + // DEBUG + stack []byte } func (tx *Tx) DBPath() string { @@ -2042,6 +2045,20 @@ func (tx *Tx) GetSortedFieldViewList() (fvs []txkey.FieldView, _ error) { return } +func (tx *Tx) DebugInfo() *TxDebugInfo { + return &TxDebugInfo{ + Ptr: fmt.Sprintf("%p", tx), + Writable: tx.writable, + Stack: string(tx.stack), + } +} + +type TxDebugInfo struct { + Ptr string `json:"ptr"` + Writable bool `json:"writable"` + Stack string `json:"stack,omitempty"` +} + // SnapshotReader returns a reader that provides a snapshot for the current database state. func (tx *Tx) SnapshotReader() (io.Reader, error) { if tx.db == nil {