Service Example 4: Custom usage counters

This example illustrates implementation of a custom usage counter in a service - for situations where a developer wants to add their own usage counters to be used in the pricing model.

The example depicts the following features:
Service execution
Dedicated service runtime implemented with ya-runtime-sdk​
Custom usage counter
Full code of the example is available in the following locations:
The requestor agent (using yapapi): https://github.com/golemfactory/yapapi/tree/b0.7/examples/custom-usage-counter​
The custom runtime implementing a custom usage counter (using ya-runtime-sdk): https://github.com/golemfactory/ya-test-runtime-counters​
Prerequisites
As with the other examples, we're assuming here you already have your yagna daemon set-up to request the test tasks and that you were able to configure your Python environment to run the examples using the latest version of yapapi. If this is your first time using Golem and yapapi, please first refer to the resources linked above.
Overview
The application consists of two components:
1.Custom runtime
A dedicated, self-contained runtime created with ya-runtime-sdk. It is a Rust binary which illustrates the API used to update the value of a custom usage counter.
2.Requestor agent
A Python application developed using yapapi. It instantiates the runtime on a Provider and periodically fetches the current state of usage counters to demonstrate fetching of custom usage counter updates.
Custom runtime implementation
The implementation of the sample runtime with a custom usage counter can be found here: https://github.com/golemfactory/ya-test-runtime-counters/blob/main/src/main.rs.
The runtime includes a metric_reporter() function which periodically increments a value of a counter named golem.usage.custom.counter, then notifies the runtime about this new value via anEventEmitter:
1
const COUNTER_NAME: &'static str = "golem.usage.custom.counter";
2
const INTERVAL: Duration = Duration::from_secs(2);
3
​
4
...
5
​
6
async fn metric_reporter(mut emitter: EventEmitter) {
7
    let mut value = 10f64;
8
​
9
    loop {
10
        tokio::time::delay_for(INTERVAL).await;
11
        value += 1f64;
12
        emitter
13
            .counter(RuntimeCounter {
14
                name: COUNTER_NAME.to_string(),
15
                value,
16
            })
17
            .await;
18
    }
19
}
Copied!
Note how RuntimeCounter struct is used to pass the value of a counter using EventEmitter's counter() method. The RuntimeCounter instance will be propagated by the ExeUnit to the Provider agent (for the purposes of pricing and invoicing) and to Requestor side where it can be fetched by the Requestor agent's code.
The remaining code includes an implementation of the ExampleRuntime, which is fairly simple, as it includes some non-void implementations of start() and run_command() actions.
In a real-world scenario, the custom counter should probably refer to some other aspect of the execution - e.g. occupied storage space or maybe the number of requests made to the service running on the provider's end.
That's because we already have two other counters that refer to time spent on execution available out of the box - one based on the wall clock time registering the time elapsed since the activity has been started (com.Counter.TIME) and another based on the actual CPU execution time (com.Counter.CPU).
start()
The start() action launches a local thread running the metric_reporter() passing anEventEmitter instance cloned from runtime's Context:
1
fn start<'a>(&mut self, ctx: &mut Context<Self>) -> OutputResponse<'a> {
2
        let emitter = match ctx.emitter.clone() {
3
            Some(emitter) => emitter,
4
            None => {
5
                let err = anyhow::anyhow!("not running in server mode");
6
                return futures::future::err(err.into()).boxed_local();
7
            }
8
        };
9
​
10
        let (handle, reg) = AbortHandle::new_pair();
11
        tokio::task::spawn_local(Abortable::new(metric_reporter(emitter.clone()), reg));
12
        self.handle = Some(handle);
13
​
14
        ...
15
    }
Copied!
This is then followed by a one-off, explicit initiation of the golem.usage.custom.counter counter value.
run_command()
The run_command() implements two explicitly named commands which can be triggered by calling a RUN <command> ExeScript call. The commands are:
sleep n- forces the runtime to wait n milliseconds.
stop- causes the runtime shutdown.
1
fn run_command<'a>(
2
        &mut self,
3
        command: RunProcess,
4
        _mode: RuntimeMode,
5
        ctx: &mut Context<Self>,
6
    ) -> ProcessIdResponse<'a> {
7
        ctx.command(|mut run_ctx| {
8
            async move {
9
                match command.bin.as_str() {
10
                    "sleep" => {
11
                        let delay_str = command
12
                            .args
13
                            .get(1)
14
                            .ok_or_else(|| anyhow::anyhow!("Missing delay value"))?;
15
​
16
                        let delay_ms: u64 = delay_str.as_str().parse()?;
17
                        let delay = Duration::from_millis(delay_ms);
18
​
19
                        run_ctx
20
                            .stdout(format!("Entering sleep for {}ms", delay_ms))
21
                            .await;
22
​
23
                        tokio::time::delay_for(delay).await;
24
                        run_ctx.stdout("Done sleeping").await;
25
                    }
26
                    "stop" => {
27
                        run_ctx.stdout("Stopping runtime").await;
28
                        run_ctx.control().shutdown();
29
                    }
30
                    _ => {
31
                        anyhow::bail!("Unsupported command {} {:?}", command.bin, command.args);
32
                    }
33
                }
34
                Ok(())
35
            }
36
            .map_err(Into::into)
37
        })
38
    }
Copied!
The example runtime implementation comes complete with ya-test-runtime-counters.json config file, which includes metadata for the runtime, required to plug it into a yagna provider service, under the name test-counters:
1
[
2
  {
3
    "name": "test-counters",
4
    "version": "0.1.0",
5
    "supervisor-path": "exe-unit",
6
    "runtime-path": "ya-test-runtime-counters/ya-test-runtime-counters",
7
    "description": "Yagna runtime supporting custom usage counters. For testing purposes only.",
8
    "extra-args": ["--runtime-managed-image"],
9
    "config": {
10
      "counters": {
11
        "golem.usage.custom.counter": {
12
          "name": "Custom",
13
          "description": "Custom counter",
14
          "price": true
15
        }
16
      }
17
    }
18
  }
19
]
Copied!
Note how config.counters structure is used to specify the custom usage counter metadata.
Pluging the runtime into golemsp
In the $HOME/.local/lib/yagna/plugins/ directory create:
file ya-test-runtime-counters.json where you describe the plugin:
1
[
2
{
3
  "name": "test-counters",
4
  "version": "0.1.0",
5
  "supervisor-path": "exe-unit",
6
  "runtime-path": "ya-test-runtime-counters/ya-test-runtime-counters",
7
  "description": "Custom usage counter example runtime",
8
  "extra-args": ["--runtime-managed-image"]
9
}
10
]
Copied!
directory ya-test-runtime-counters (compare runtime-path in above file) where ya-test-runtime-counters binary along with Erigon binaries are placed.
New runtime needs also to be enabled in $HOME/.local/share/ya-provider/presets.json. Preset object can be copied from other presets. Please note that exeunit-name has to match to the name property of the plugin above:
1
{
2
  "active": [
3
    "test-counters",
4
    ...
5
  ],
6
  "presets": [
7
    {
8
      "name": "test-counters",
9
      "exeunit-name": "test-counters",
10
      "pricing-model": "linear",
11
      "initial-price": 0,
12
      "usage-coeffs": {
13
        "golem.usage.duration_sec": 0.0001,
14
        "golem.usage.cpu_sec": 0.0001,
15
        "golem.usage.custom.counter": 0.0003
16
      }
17
    },
Copied!
Note how in the example above the golem.usage.custom.counter is to be included in the pricing function (linear model) with a coefficient of 0.0003 GLM.
Requestor agent
The requestor agent is a fairly simple implemenation of a Golem service which:
Requires a test-counters runtime as payload:
1
@dataclass
2
class CustomCounterServicePayload(Payload):
3
    runtime: str = constraint(inf.INF_RUNTIME_NAME, default="test-counters")
Copied!
In the run() handler, it periodically fetches the usage vector as published by the ExeUnit alongside the accumulated cost and displays it in console:
1
async def run(self):
2
    start_time = datetime.now()
3
    print(f"service {self.id} running on '{self.provider_name}'...")
4
    while datetime.now() < start_time + timedelta(seconds=self._running_time_sec):
5
        script = self._ctx.new_script()
6
        script.run("sleep", "1000")
7
        yield script
8
        usage: ActivityUsage = await self._ctx.get_usage()
9
        cost = await self._ctx.get_cost()
10
        print(f"total cost so far: {cost}; activity usage: {usage.current_usage}")
11
        await asyncio.sleep(3)
Copied!
Launching the service
To launch the service, we first need to initialize Golem:
1
async with Golem(
2
        budget=10.0, subnet_tag=subnet_tag, driver=driver, network=network, strategy=strategy
3
    ) as golem:
4
        instance_params = [{"running_time_sec": running_time_sec}]
5
        
Copied!
Once created, we use it to instantiate a CustomCounterService:
1
cluster = await golem.run_service(CustomCounterService, instance_params=instance_params)
Copied!
And that's all there is to it. Running the example should result with a sequence of messages appearing in console, showing the golem.usage.custom.countercounter being periodically incremented.