| 469 | "sessions": h.pool.ActiveSessions(), |
| 470 | "active_queries": h.pool.ActiveDrainWork(), |
| 471 | "uptime_ns": time.Since(h.pool.startTime).Nanoseconds(), |
| 472 | "session_progress": sessionProgress, |
| 473 | "instance_invalidated": instanceInvalidated, |
| 474 | "instance_invalid_reason": h.pool.InstanceInvalidReason(), |
| 475 | "otlp_export_enabled": cliboot.OTLPExportEnabled(), |
| 476 | "otlp_export_failures": cliboot.OTLPExportFailures(), |
| 477 | }) |
| 478 | return sendActionResult(stream, &flight.Result{Body: resp}) |
| 479 | } |
| 480 | |
| 481 | func (h *FlightSQLHandler) doWaitSessionIdle(body []byte, stream flight.FlightService_DoActionServer) error { |
| 482 | var req server.WorkerWaitSessionIdlePayload |
| 483 | if err := json.Unmarshal(body, &req); err != nil { |
| 484 | return status.Errorf(codes.InvalidArgument, "invalid WaitSessionIdle request: %v", err) |
| 485 | } |
| 486 | if err := h.pool.validateControlMetadata(req.WorkerControlMetadata); err != nil { |
| 487 | return status.Errorf(codes.FailedPrecondition, "stale worker owner: %v", err) |
| 488 | } |
| 489 | session, err := h.sessionFromContext(stream.Context()) |
| 490 | if err != nil { |
| 491 | return err |
| 492 | } |
| 493 | if err := session.waitOperationIdle(stream.Context()); err != nil { |
| 494 | switch { |
| 495 | case errors.Is(err, context.Canceled): |
| 496 | return status.Errorf(codes.Canceled, "wait for session idle: %v", err) |
| 497 | case errors.Is(err, context.DeadlineExceeded): |