File size: 4,331 Bytes
a00522e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """Minimal bucket mount probe for Gradio SDK Spaces.
Runs the probe once at import time (so results appear in startup logs even if
the Gradio UI fails to load), and also exposes it via a Gradio interface.
"""
import os
import pwd
import sqlite3
import subprocess
import traceback
try:
import fcntl
except ImportError:
fcntl = None
import gradio as gr
def _run_probe() -> str:
lines = []
lines.append("=" * 60)
lines.append("BUCKET MOUNT PROBE — Gradio SDK Space")
lines.append("=" * 60)
# Runtime identity
try:
uid = os.getuid()
user = pwd.getpwuid(uid).pw_name
lines.append(f"uid={uid} user={user} euid={os.geteuid()} gid={os.getgid()}")
except Exception as e:
lines.append(f"uid/user lookup failed: {e}")
# Env hints
lines.append(f"SYSTEM={os.environ.get('SYSTEM')!r}")
lines.append(f"SPACE_ID={os.environ.get('SPACE_ID')!r}")
lines.append(f"HOME={os.environ.get('HOME')!r}")
# Mount listing
lines.append("")
lines.append("--- ls -lan /data ---")
try:
out = subprocess.run(
["ls", "-lan", "/data"], capture_output=True, text=True, timeout=10
)
lines.append(out.stdout or "(empty)")
if out.stderr:
lines.append(f"stderr: {out.stderr}")
except Exception as e:
lines.append(f"ls failed: {e}")
lines.append("--- stat /data ---")
try:
out = subprocess.run(
["stat", "/data"], capture_output=True, text=True, timeout=10
)
lines.append(out.stdout or "(empty)")
except Exception as e:
lines.append(f"stat failed: {e}")
lines.append("--- mount | grep /data ---")
try:
out = subprocess.run(
"mount | grep /data || true",
shell=True,
capture_output=True,
text=True,
timeout=10,
)
lines.append(out.stdout or "(no match)")
except Exception as e:
lines.append(f"mount failed: {e}")
# Write probes
lines.append("")
lines.append("--- write probes ---")
def probe(label: str, fn):
try:
fn()
lines.append(f"OK {label}")
except Exception as e:
lines.append(f"FAIL {label}: {type(e).__name__}: {e}")
probe(
"touch /data/gradio_probe_touch",
lambda: open("/data/gradio_probe_touch", "w").close(),
)
def _sqlite_create():
conn = sqlite3.connect("/data/gradio_probe.db", timeout=5.0)
conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
conn.execute("INSERT INTO t VALUES (1)")
conn.commit()
conn.close()
probe("sqlite3 connect + CREATE + INSERT", _sqlite_create)
def _sqlite_delete_journal():
# This is what trackio does on Spaces: journal_mode=DELETE instead of WAL
conn = sqlite3.connect("/data/gradio_probe_delete.db", timeout=5.0)
conn.execute("PRAGMA journal_mode = DELETE")
conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
conn.commit()
conn.close()
probe("sqlite3 journal_mode=DELETE", _sqlite_delete_journal)
if fcntl is not None:
def _flock():
f = open("/data/gradio_probe.lock", "w")
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
f.close()
probe("fcntl.flock LOCK_EX|LOCK_NB", _flock)
else:
lines.append("SKIP fcntl.flock: fcntl not importable")
lines.append("=" * 60)
return "\n".join(lines)
# Run once at import time so results land in startup logs
try:
_STARTUP_PROBE = _run_probe()
except Exception:
_STARTUP_PROBE = "startup probe crashed:\n" + traceback.format_exc()
print(_STARTUP_PROBE, flush=True)
def probe_handler():
return _run_probe()
with gr.Blocks() as demo:
gr.Markdown("# Bucket mount probe — Gradio SDK Space")
gr.Markdown(
"Attach a Storage Bucket at `/data` via Space settings (Read/Write), "
"then click the button below. Startup-time probe output is in the Space logs."
)
with gr.Row():
btn = gr.Button("Run probe")
out = gr.Textbox(label="Probe output", lines=30, value=_STARTUP_PROBE)
btn.click(fn=probe_handler, inputs=[], outputs=out)
demo.launch()
|