Monitoring and Observability

s = index.describe_index_stats()
print(s["total_vector_count"], s["dimension"], s.get("index_fullness"))

for ns, info in s["namespaces"].items():
    push_metric("pinecone.ns.vector_count",
                value=info["vector_count"], tags={"ns": ns})

from prometheus_client import Histogram
qlat = Histogram("pinecone_query_latency_seconds", "...")

with qlat.time():
    res = index.query(vector=qv, top_k=10)

from prometheus_client import Counter
err = Counter("pinecone_errors_total", "...", ["op", "code"])

try:
    index.upsert(vectors=batch)
except Exception as e:
    err.labels(op="upsert", code=getattr(e, "status", "unknown")).inc()
    raise

import logging, time
log = logging.getLogger("pinecone")

def query_logged(qv, top_k):
    t = time.perf_counter()
    res = index.query(vector=qv, top_k=top_k)
    log.info("query", extra={"top_k": top_k,
        "matches": len(res["matches"]),
        "ms": (time.perf_counter()-t)*1000})
    return res