Service examples

The Storage API allows data exposure from non-standard storage backends to services using the Storage API. For instance, local NAS data or an on-premise database can be made available to clients via a bespoke Storage API instance implementing the specified functions for that particular backend.

REST service instance example

This section demonstrates a simple REST service for the Storage API using Python and FastAPI.

Implementing stat

First, install fastapi, pydantic, asyncio, and uvicorn:

pip install fastapi pydantic uvicorn asyncio

Next, implement a service offering the stat endpoint, serving content of a local disk directory via the Storage API. We assume the directory to be served is stored in the global variable STATIC_DIR:

@app.head("/fileobject/by-address/{resource_address:path}")
async def stat_api(resource_address: str):
    path = os.path.join(STATIC_DIR, resource_address)
    if os.path.exists(path):
        size = os.path.getsize(path)
        modification_time = datetime.fromtimestamp(os.path.getmtime(path), tz=timezone.utc).isoformat()
        metadata = Metadata(data_object_size=size, last_modified_timestamp=modification_time)
        return JSONResponse(
            status_code=204,
            content=None,
            headers={
                "x-nvidia-omniverse-storage-metadata": metadata.model_dump_json(),
                "x-nvidia-omniverse-storage-resource-identity": create_identity(resource_address, modification_time),
            },
        )
    else:
        raise HTTPException(status_code=404, detail=f"{path} not found on disk")

The code checks if the path exists, raises an HTTP 404 if not, gathers required metadata, and builds a JSON response.

Pydantic mapping is used for automatic serialization and type checking:

class Metadata(BaseModel):
    data_object_size: int
    last_modified_timestamp: Optional[str] = None

The resource identity is defined as the combination of the address (relative file path) and modification date, encoded as a JSON data package and encoded using base64. This is a design choice to emphasize that the resource identity’s format is storage service implementation specific and the client should make no assumption about its content.

def create_identity(relative_path: str, modification_time: str) -> str:
    return binascii.b2a_base64(json.dumps({"p": relative_path, "t": modification_time}).encode("utf-8")).decode("utf-8").strip()

Running the service example is done with uvicorn:

    server = grpc.server(concurrent.futures.ThreadPoolExecutor(max_workers=10))
    add_FileObjectServiceServicer_to_server(FileSystemServiceServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print(f"GRPC Server listening on port 50051 serving directory '{STATIC_DIR}'")
    server.wait_for_termination()

Test the service with curl, here we have placed a file hello.txt in the STATIC_DIR being served:

> curl --head localhost:8011/object-store/by-address/hello.txt
HTTP/1.1 204 No Content
date: Fri, 28 Jun 2024 13:17:37 GMT
server: uvicorn
x-nvidia-omniverse-storage-metadata: {"resource_address":"hello.txt","resource_identity":"eyJwIjogImhlbGxvLnR4dCIsIC
J0IjogIjIwMjQtMDYtMjhUMTM6MTI6MzUuNzY3MTM4KzAwOjAwIn0=",
"metadata":{"data_object_size":12,"last_modified_timestamp":"2024-06-28T13:12:35.767138+00:00"}}
content-type: application/json

Direct download implementation

The read operation streams the file back via the HTTP response body:

@app.get("/fileobject/by-identity/{resource_identity:path}")
async def read(resource_identity: str, download_preference: Optional[str] = None):
    relative_path = path_from_identity(urllib.parse.unquote_plus(resource_identity))
    path = os.path.join(STATIC_DIR, relative_path)
    if os.path.exists(path):

        def load_file():
            with open(path, "rb") as f:
                while chunk := f.read(1024):
                    yield chunk

        return StreamingResponse(load_file(), media_type="application/octet-stream")
    else:
        raise HTTPException(status_code=404, detail=f"{path} not found on disk")

Use the previous stat operation result to download the file content from the Storage API service:

curl --get localhost:8011/object-store/by-identity/eyJwIjogImhlbGxvLnR4dCIsICJ0IjogIjIwMjQtMDYtMjhUMTM6MTI6MzUuNzY3MTM4KzAwOjAwIn0= -v
*   Trying [::1]:8011...
*   Trying 127.0.0.1:8011...
* Connected to localhost (127.0.0.1) port 8011
> GET /object-store/by-identity/eyJwIjogImhlbGxvLnR4dCIsICJ0IjogIjIwMjQtMDYtMjhUMTM6MTI6MzUuNzY3MTM4KzAwOjAwIn0= HTTP/1.1
> Host: localhost:8011
> User-Agent: curl/8.4.0
> Accept: */*
>
< HTTP/1.1 200 OK
< date: Fri, 28 Jun 2024 13:32:37 GMT
< server: uvicorn
< content-type: application/octet-stream
< transfer-encoding: chunked
<
Hello from your demo server, this is the content of the file hello.txt!
* Connection #0 to host localhost left intact

gRPC service example

For the gRPC service, generate Python language bindings from the proto files. The client example section contains detailed information on that.

gRPC Stat

The gRPC service is implemented by subclassing the base class and implementing the methods:

from nvidia.omniverse.storage.fileobject.v1alpha.fileobject_service_pb2_grpc import (
    FileObjectServiceServicer,
    add_FileObjectServiceServicer_to_server,
)
from service_utils import (
    STATIC_DIR,
    args,
    create_identity,
)


class FileSystemServiceServicer(FileObjectServiceServicer):
    def Stat(self, request, context):
        path = os.path.join(STATIC_DIR, request.resource_address)
        if os.path.exists(path):
            # First build the ResourceIdentity from resource address and modification time
            modification_time = os.path.getmtime(path)
            modification_time_iso = datetime.fromtimestamp(modification_time, tz=timezone.utc).isoformat()
            resource_identity = fileobject_pb2.ResourceIdentity(
                encoded_identity=create_identity(request.resource_address, modification_time_iso)
            )

            # Then create the Metadata info using google Timestamp
            stat_result = os.stat(path)
            modification_time_seconds = int(stat_result.st_mtime)
            nanos = int((stat_result.st_mtime - modification_time_seconds) * 1e9)
            timestamp = timestamp_pb2.Timestamp(seconds=int(stat_result.st_mtime), nanos=nanos)
            metadata = fileobject_pb2.Metadata(data_object_size=stat_result.st_size, last_modified_timestamp=timestamp)

            # Now return the result
            return fileobject_service_pb2.StatResponse(
                resource_info=fileobject_pb2.ResourceInfo(resource_identity=resource_identity, metadata=metadata)
            )
        else:
            context.abort(grpc.StatusCode.NOT_FOUND, f"file not found: {path}")

The server main loop is:

    server = grpc.server(concurrent.futures.ThreadPoolExecutor(max_workers=10))
    add_FileObjectServiceServicer_to_server(FileSystemServiceServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print(f"GRPC Server listening on port 50051 serving directory '{STATIC_DIR}'")
    server.wait_for_termination()

Run the server to get a working gRPC server in plaintext mode, queryable via grpcurl for example:

grpcurl -plaintext -d "{\"resource_address\": \"hello.txt\"}" -import-path c:\src\test_impl\storage-protos\proto\nvidia\omniverse\storage\fileobject\v1beta -import-path c:\src\test_impl\storage-protos\proto -proto fileobject_service.proto localhost:50051 nvidia.omniverse.storage.fileobject.v1beta.FileObjectService.Stat

The expected reply is:

{
  "resourceInfo": {
    "resourceIdentity": {
      "encodedIdentity": "eyJwIjogImhlbGxvLnR4dCIsICJ0IjogIjIwMjQtMDYtMjhUMTM6MTI6MzUuNzY3MTM4KzAwOjAwIn0="
    },
    "metadata": {
      "dataObjectSize": "12",
      "lastModifiedTimestamp": "2024-06-28T13:12:35.767138242Z"
    }
  }
}

gRPC reflection

Enable reflection in the gRPC server to simplify the grpcurl command. Install the package:

pip install grpcio-reflection

Modify the gRPC server launch sequence to include the reflection extension:

    server = grpc.server(concurrent.futures.ThreadPoolExecutor(max_workers=10))
    add_FileObjectServiceServicer_to_server(FileSystemServiceServicer(), server)

    # Enable reflection
    from grpc_reflection.v1alpha import reflection

    SERVICE_NAMES = (
        nvidia.omniverse.storage.fileobject.v1alpha.fileobject_service_pb2.DESCRIPTOR.services_by_name["FileObjectService"].full_name,
        reflection.SERVICE_NAME,
    )
    reflection.enable_server_reflection(SERVICE_NAMES, server)
    server.add_insecure_port("[::]:50051")
    server.start()
    print(f"GRPC Server listening on port 50051 serving directory '{STATIC_DIR}'")
    server.wait_for_termination()

Relaunch the server, query the enabled services:

> grpcurl -plaintext localhost:50051 list
grpc.reflection.v1beta.ServerReflection
nvidia.omniverse.storage.fileobject.v1beta.FileObjectService

Get more information on the individual functions:

> grpcurl -plaintext localhost:50051 describe nvidia.omniverse.storage.fileobject.v1beta.FileObjectService
nvidia.omniverse.storage.fileobject.v1beta.FileObjectService is a service:
service FileObjectService {
  rpc Stat ( .nvidia.omniverse.storage.fileobject.v1beta.StatRequest ) returns ( .nvidia.omniverse.storage.fileobject.v1beta.StatResponse );
}

Call the Stat function:

> grpcurl -plaintext -d "{\"resource_address\": \"hello.txt\"}" localhost:50051 nvidia.omniverse.storage.fileobject.v1beta.FileObjectService.Stat