Skip to content

Tutorials

Jump to bottom
Álvaro Puente edited this page Oct 23, 2025 · 16 revisions

App Requirements description

To enable the user application to define application requirements, also known as functional requirements, it should allow the user to specify preferences for the orchestration of resources where the function will be offloaded. This will empower users to customize and optimize the resource allocation based on their specific needs.

These requirements should match with the data model define in the following file: data model

The meaning of the available requirements is the following:

  • FLAVOUR: It is a string that describes the flavour of the runtime. It is mandatory to set this requirement as it is used to define in which type of serverless runtime will the functions be executed in terms of dependencies and computing resources.
  • MAX_LATENCY: It defines the maximun latency that the user application will tolerate when offloading a function. It is an integer optional parameter that, when it is established, it will select the Edge Cluster with the least latency. Therefore, the COGNIT framework is currently a best effort service.
  • MAX_FUNCTION_EXECUTION_TIME: It is a float optional requirement that sets the max execution time allowed for the offloaded function.
  • MIN_ENERGY_RENEWABLE_USAGE: An integer optional requirement that sets the minimum energy renewable percentage to be used when executing the function.
  • GEOLOCATION: It is a mandatory requirement as it is used to select the nearest Edge Cluster for the user application if the MAX_LATENCY requirement is not set. This parameter is composed of a dictionary with two keys ("latitude" and "longitude") that have to be filled with float values.
  • ID: Unique identifier assigned to each device. Used to estimate load and improve optimizer decisions. This field is mandatory and must be of type string.
  • PROVIDER: It is an optional field that provides a list of provider identifiers used to restrict cluster selection by provider.
  • IS_CONFIDENTIAL: It is an optional requirement and indicates if Confidential Computing is required for the execution of the functions. By default this requirement is set to false.

The way to set these requirements is demonstrated in this example

Offload MinIO example

Before going deep on what COGNIT offers in regard to MinIO, it's important to note that MinIO does not differentiate between files and folders: everything inside a bucket is considered an object.

However, externally MinIO does support a linux like estructure using slash ("/"). To understand this difference, consider these two examples:

  1. Save a file to the root bucket path:

    minio_client.upload_object(
    bucket="ikerlan",
    objectPath="testData.csv",
    data=data
)

    Result inside MinIO:

    ikerlan(bucket)
└── testData.csv

  2. Save a file inside a folder:

    minio_client.upload_object(
    bucket="ikerlan",
    objectPath="recordings/20250512/recording1.csv",
    data=data
)

    Result inside MinIO:

    ikerlan(bucket)
└── recordings
    └── 20250512
        └── recording1.csv

In order to enable communication between an offloaded function and the MinIO service, a Minio Client has been developed within the Serverless Runtime. This client provides a set of methods to interact with the service. As for bucket-oriented methods it has:

  • create_bucket(bucket_name)

    • Create a bucket.

  • list_buckets() -> List[str]

    • List all buckets.

  • delete_bucket(bucket_name)

    • Delete an empty bucket.

  • enable_versioning(bucket_name)

    • Enable versioning for a bucket.

  • set_bucket_policy(bucket_name, policy)

    • Set a JSON policy document on a bucket.

While object-based methods the Minio Client has:

  • list_objects(bucket_name) -> List[str]

    • List all object keys in a bucket.

  • upload_object(bucket, objectPath, data, extraArgs=None) -> str

    • Upload raw bytes (data) to bucket at key objectPath, with optional extraArgs (metadata, ACL).

  • download_object(bucket: str, key: str, download_path: str = None) -> Union[bytes, str, Exception]

    • If download_path is provided, save the object to that Serverless Runtime path and return the path; otherwise return raw bytes.

    WARNING: If download_path is provided, the downloaded file stays in the SR, meaning that no data is transferred back to the DR.

  • delete_object(bucket_name, object_name)

    • Delete a single object from a bucket.

  • get_object_metadata(bucket_name, object_name) -> dict

    • Retrieve the user-defined metadata for an object.

  • copy_object(source_bucket, source_key, destination_bucket, destination_key)

    • Copy an object from one bucket/key to another bucket/key.

There are also Serverless Runtime local disk methods:

  • download_file_to_sr_disk(bucket_name, object_name, download_path)

    • Download a single file from a bucket to the Serverless Runtime’s local disk at download_path.

  • download_objects_with_prefix_to_sr_disk(bucket_name, prefix, target_local_directory, preserve_nested_structure=False)

    • Download all objects in bucket_name whose keys start with prefix into target_local_directory.
      • If preserve_nested_structure is False, files are flattened into target_local_directory.
      • If True, the full key path under prefix is recreated under target_local_directory.

  • download_object(bucket: str, key: str, download_path: str = None) -> Union[bytes, str, Exception]

    • For the cases that download_path is provided, as it is explained in "Object level methods" section.

Function offloading guidelines

To use MinIO within a function that is offloaded to the serverless runtime, you should import and use the MinIO client provided by the Serverless Runtime library directly inside the function. The sentence that imports the client withing the function is:

from modules._minio_client import MinioClient

So the MinIO client can connect to the MinIO service, the following parameters must be provided when initializing the MinIO client:

  • endpoint_url
    • String defining the Minio endpoint.
    • It must have the format "http://IP:PORT" (Important: "http://" must be included).
    • Note: Not tested with HTTPS.
  • access_key
    • String defining the Minio username.
  • secret_key
    • String defining the Minio password.

An example of a function to be offloaded using MinIO is the following:

def offload_function_example():
        from modules._minio_client import MinioClient
        minio_client = MinioClient(
            endpoint_url=MINIO_ENDPOINT,
            access_key="minio_user",
            secret_key="minio_psw"
        )
        response = minio_client.list_buckets()
        return response

Additionally, you can find a broader example here

Tip

Since the function is offloaded from the Serverless Runtime, it is the Serverless Runtime that must connect to the MinIO service. Please ensure that the Serverless Runtime has network access to the specified endpoint_url.

Enabling offloaded function logs in the SR log

There is a possibility to log the internal state of the offloaded funcions inside the Serverless Runtime app log. To do so, the user will need to include the Serverless Runtime logger class inside of each offloaded function that wants to log within the SR log:

from modules._logger import CognitLogger
cognit_logger = CognitLogger()

Afterwards the user can use the log levels:

cognit_logger.debug("This is a     DEBUG     level message")
cognit_logger.info("This is an     INFO      level message")
cognit_logger.warning("This is a   WARNING   level message")
cognit_logger.error("This is an    ERROR     level message")
cognit_logger.critical("This is a  CRITICAL  level message")

When the SR app runs the offloaded function, the logger will also include the logs registered within that function. An example of log registrations can be found here

Geolocation based on IP address service

To obtain a device's geolocation coordinates based on its IP address, you can use ipinfo.io . Signing up is required to obtain a token, which allows access to certain restricted features of the RESTful API.

Once the user has registered and obtained a token, the following script demonstrates how to retrieve geolocation information that can be used in a given example:

It uses both restricted and unrestricted endpoints to fetch the device's public IP address and retrieve geolocation information based on it.

import requests

def get():
    endpoint = 'https://ipinfo.io/ip'
    response = requests.get(endpoint, verify = True)

    if response.status_code != 200:
        return 'Status:', response.status_code, 'Problem with the request. Exiting.'
        exit()

    #data = response.json()
    data = response.text.strip()

    #return data['ip']
    return data

token="<your_token_value_here>"
my_addr=get()
endp = "https://ipinfo.io/{0}?token={1}".format(my_addr, token)
#get my ip
r = requests.get(endp, verify = True)

#print my geoloc
print(r.text)

Output of the example script that can be parsed to use as COGNIT's geolocation coordinate input:

{
  "ip": "XXX.XXX.XXX.XXX",
  "city": "Your city",
  "region": "Basque Country",
  "country": "ES",
  "loc": "XX.xx,YY.yy",
  "org": "XX Autonomous System",
  "postal": "20500",
  "timezone": "Europe/Madrid"
}

Clone this wiki locally