Metering collection tool

All Corda Enterprise nodes record metering data, the metering collection tool is used to collect data from the node. Node metering data can also be collected using the metering client.

The metering collection tool provides a mechanism for collecting metering data from nodes and notaries running Corda Enterprise. The metering collection tool is a CorDapp that includes a number of flows which return data from the node.

Corda metering data

The Corda Enterprise metering process is based on the signing of transactions.

Once a transaction is signed, it cannot be modified without invalidating this signature. Each transaction signature is a signing event. The node records a small piece of data whenever a signing event occurs. This data describes which entity signed the transaction, what CorDapps and commands were involved, and the time when the signing event occurred.

Notaries running on Corda Enterprise are also metered. The data recorded for notaries indicates what notarisation requests were made and who made them.

How metering data is shared

The metering collection tool also contains responder flows that can be used by other nodes on the network to collect metering data from the node where the respective CorDapp is installed. This feature must be enabled by the node operator deploying a CorDapp configuration file for the CorDapp.

If no configuration file is deployed, metering data will not be shared with any other network party. An example configuration file that enables metering data sharing is shown below:

"access_configuration" : {
    "network_collectors" : ["O=PartyA,L=New York,C=US", "O=PartyB,L=Zurich,C=CH"],
    "cordapp_collectors" : {
        "by_name" : {
            "Corda Finance Demo" : ["O=PartyB,L=Zurich,C=CH"]
        },
        "by_hash" : {
          "FC0150EFAB3BBD715BDAA7F67B4C4DB5E133D919B6860A3D3B4C6C7D3EFE25D5" :
            ["O=PartyC,L=London,C=GB"],
          "44489E8918D7D8F7A3227FE56EC34BFDDF15BD413FF92F23E72DD5D543BD6194" :
            ["O=PartyC,L=London,C=GB"]
        },
        "by_signature" : {
          "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" :
            ["O=PartyD,L=Dublin,C=IE"]
        }
    }
}

Based on the example configuration above:

• Nodes PartyA and PartyB collect aggregated metering data from the node. This means that only the total number of signing events, which have happened within the specified time period, are shared.
• Node PartyB node collects detailed metering data related to all installed CorDapps called Corda Finance Demo (the name must be an exact match).
• Node PartyC collects detailed metering data related to CorDapps with a JAR hash either FC0150EFAB3BBD715BDAA7F67B4C4DB5E133D919B6860A3D3B4C6C7D3EFE25D5 or 44489E8918D7D8F7A3227FE56EC34BFDDF15BD413FF92F23E72DD5D543BD6194.
• Node PartyD collects detailed metering data related to all CorDapps that have had their JAR files signed with the key AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B.

To create the configuration file correctly, use the RetrieveCordappDataFlow flow to get detailed information about the CorDapps deployed on your node.

• Use the RetrieveCordappDataFlow flow to get detailed information about the CorDapps deployed on your node.
• Ensure you configure the correct values for the configuration file static keys (access_configuration, network_collectors, by_hash, and so on). Any errors, like a typo, will mean your configuration is ignored and the default applied. As a result, no metering data will be shared.
• Ensure that every JAR hash, JAR signature, and CorDapp name in the configuration matches at least one of the deployed CorDapps. This means that you must not whitelist a CorDapp that does not exist. This step is essential in order to pass the configuration validation step that runs at node start-up, which checks that the X.500 names used in the configuration file are valid. If the configuration validation step fails for any reason, the node will fail to start.

Install the metering collection tool

The metering collection tool is distributed as part of Corda Enterprise Edition 4.10 under the name corda-tools-metering-collector-4.10.jar. You must place this JAR file in the cordapps directory of the node.

Use the metering collection flows

The metering collection tool is a CorDapp containing a number of flows that return meting data from a node or notary. The flows are:

• MeteringCollectionFlow
• NodeMeteringCollectionFlow
• FilteredMeteringCollectionFlow
• AggregatedMeteringCollectionFlow
• MultiFilteredCollectionFlow
• MultiAggregatedCollectionFlow
• NotaryCollectionFlow
• RetrieveCordappDataFlow

MeteringCollectionFlow

Use the MeteringCollectionFlow to collect metering data from a node. You must specify the time window over which to collect the metering data, and can also specify CorDapps to filter data by.

The flow returns the total count of metering events that match the filters within the time window, and a breakdown of these events based on the commands involved and the signing entities based on the information held by the current node.

You invoke this flow from the shell. The flow takes the following arguments:

• A time window over which the flow runs. This is a mandatory argument. The accepted time window formats are either a start date and an end date (both of type Instant), or a start date and a duration (see the Usage section below). Note that the minimum time unit you can use is an hour, so the flow is unable to collect metering data over durations shorter than an hour.
• A filter to select which CorDapps to collect data for. To specify a filter, provide a MeteringFilter object, which consists of filterBy criteria and a list of strings that describe the CorDapps to filter by. There are four possible options to filter by, which are described in the data filtering section.
• A paging specification to describe how the flow should access the database. The paging specification is used to control database access by ensuring that only a subset of data is accessed at once. This is important in order to prevent too much data being read into memory at once, which would result in out-of-memory errors. By default, up to 10 000 metering entries are read into memory at a time, although the number of returned entries is likely to be smaller because some aggregation takes place in the background. If more than one page of data is required, the flow may need to be run multiple times to collect the full breakdown of metering events. However, the total count provided is always the full number of signing events that match the supplied criteria.

Use the shell interface to invoke the flow by specifying the time window - either provide the startDate and endDate for the metering data collection in the format YYYY-MM-DD, or the startDate (in the same format) and the duration as an integer number of daysToCollect.

You can also specify a filter according to the rules described in the data filtering section. This is not needed if all the metering data is required. As mentioned above, the smallest time window you can specify is one day.

When date strings are required, they are always in the YYYY-MM-DD format. If the date does not parse correctly, an exception is thrown.

The example below shows a collection of all metering data over a particular week:

start MeteringCollectionFlow startDate: 2019-11-07, daysToCollect: 7, page: 1

The example below shows a collection of metering data for a particular CorDapp:

start MeteringCollectionFlow startDate: 2019-11-07, endDate: 2019-11-14, filterBy: CORDAPP_NAMES, filter: ["Corda Finance Demo"], page: 1

Output

The output of the MeteringCollectionFlow flow is a data class that contains a structured representation of the metering data, as follows:

• The total number of signing events that match the query provided.
• The current version of the output metering data.
• An object describing the query that produced this set of data. This includes the time window over which the data was collected, the filter applied to the data, and the paging criteria used.
• A list of entries giving a breakdown of the metering data. Each entry contains a signing entity, a set of commands, a transaction type, and a count of events in this page that match this specification.

The output object can also be serialised into JSON format by calling serialize.

When you run metering collection tool from the shell, the collected metering data is shown as output on the shell terminal, in JSON format. The example below shows the output JSON on the shell terminal:

{"totalCount":2,"version":1,"query":{"startDate":"2019-11-13T00:00:00Z","endDate":"2019-11-15T00:00:00Z","filter":{"filterBy":"NONE","values":[]},"pageNumber":1,"totalPages":1,"pageSize":10000},"entries":[{"signingId":{"type":"NODE_IDENTITY","accountId":null},"txType":"NORMAL","commands":["net.corda.finance.contracts.asset.Cash.Commands.Issue"],"count":1},{"signingId":{"type":"NODE_IDENTITY","accountId":null},"txType":"NORMAL","commands":["net.corda.finance.contracts.asset.Cash.Commands.Move"],"count":1}]}

NodeMeteringCollectionFlow

Use the NodeMeteringCollectionFlow flow to collect metering data from a node by connecting to it via RPC. This flow functions in the same manner as MeteringCollectionFlow and returns the same information, but this flow can be invoked via RPC rather than the node shell.

You can use this flow to retrieve detailed metering data from the node the RPC client connects to. The flow is similar to the MeteringCollectionFlow flow, however it provides a simpler API that does not require data pagination.

The example below shows how to retrieve metering data from a node running on the local machine, for CorDapps named myCorDapp1 and myCorDapp2 (as well as any other CorDapp that contains those two strings in its name), for the duration of the past 7 days:

NetworkHostAndPort hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000");
CordaRPCClient client = new CordaRPCClient(hostAndPort);
NodeMeteringData meteringData =
        client.use("rpcUsername", "rpcPassword", conn -> {
            CordaRPCOps rpcOps = conn.getProxy();
            Instant now = Instant.now();
            Instant sevenDaysAgo = now.minus(7, ChronoUnit.DAYS);
            FlowHandle<NodeMeteringData> handle = rpcOps.startFlowDynamic(
                    NodeMeteringCollectionFlow.class,
                    new Filter.And(
                            Filter.ByTimeStamp.between(sevenDaysAgo, now),
                            new Filter.Or(
                                    new Filter.ByCordapp.ByName("myCorDapp1"),
                                    new Filter.ByCordapp.ByName("myCorDapp2")
                            )
                    ));
            Future<NodeMeteringData> result = handle.getReturnValue();
            try {
                return result.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

val hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000")
val client = CordaRPCClient(hostAndPort)
val nodeMeteringData = client.use("rpcUsername", "rpcPassword") { conn: CordaRPCConnection ->
    val rpcOps = conn.proxy
    val now = Instant.now()
    val sevenDaysAgo = now.minus(7, ChronoUnit.DAYS)
    val handle = rpcOps.startFlow(
            ::NodeMeteringCollectionFlow, Filter.And(
            Filter.ByTimeStamp.between(sevenDaysAgo, now),
            Filter.Or(Filter.ByCordapp.ByName("myCorDapp1"), Filter.ByCordapp.ByName("myCorDapp2"))
    ))
    val result: Future<NodeMeteringData> = handle.returnValue
    try {
        result.get()
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}

FilteredMeteringCollectionFlow

Use the FilteredMeteringCollectionFlow flow to collect metering data from another node on the network by connecting to it via RPC. The FilteredMeteringCollectionFlow flow is identical to the NodeMeteringCollectionFlow flow except that FilteredMeteringCollectionFlow collects data from another node on the network. When using FilteredMeteringCollectionFlow, you must specify the party running the node to collect data from.

You can use this flow to collect metering data from a remote node on the network. It is similar to the NodeMeteringCollectionFlow flow - the difference between the two is that FilteredMeteringCollectionFlow collects metering data from a remote node on the network and NodeMeteringCollectionFlow collects metering data from the node the RPC client connects to.

The example below shows how to retrieve metering related to CorDapps with names myCorDapp1 and myCorDapp2 (as well as any other CorDapp with a name containing any of those two strings), connecting to a node running on the local machine, from the node ran by O=PartyA,L=New York,C=US, for the duration of the past 7 days:

NetworkHostAndPort hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000");
CordaRPCClient client = new CordaRPCClient(hostAndPort);
FilteredNodeMeteringData meteringData =
        client.use("rpcUsername", "rpcPassword", conn -> {
            CordaRPCOps rpcOps = conn.getProxy();
            Party destination = rpcOps.wellKnownPartyFromX500Name(
                    CordaX500Name.parse("O=PartyA,L=New York,C=US")
            );
            Instant now = Instant.now();
            Instant sevenDaysAgo = now.minus(7, ChronoUnit.DAYS);
            FlowHandle<FilteredNodeMeteringData> handle = rpcOps.startFlowDynamic(
                    FilteredMeteringCollectionFlow.class,
                    destination,
                    new Filter.And(
                            Filter.ByTimeStamp.between(sevenDaysAgo, now),
                            new Filter.Or(
                                    new Filter.ByCordapp.ByName("myCorDapp1"),
                                    new Filter.ByCordapp.ByName("myCorDapp2")
                            )
                    )
            );
            Future<FilteredNodeMeteringData> result = handle.getReturnValue();
            try {
                return result.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

val hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000")
val client = CordaRPCClient(hostAndPort)
val data = client.use("rpcUsername", "rpcPassword") { conn: CordaRPCConnection ->
    val rpcOps = conn.proxy
    val destination = rpcOps.wellKnownPartyFromX500Name(
            CordaX500Name.parse("O=PartyA,L=New York,C=US")
    )!!
    val now = Instant.now()
    val sevenDaysAgo = now.minus(7, ChronoUnit.DAYS)
    val handle = rpcOps.startFlow(
            ::FilteredMeteringCollectionFlow,
            destination,
            Filter.And(
                    Filter.ByTimeStamp.between(sevenDaysAgo, now),
                    Filter.Or(
                            Filter.ByCordapp.ByName("myCorDapp1"),
                            Filter.ByCordapp.ByName("myCorDapp2")
                    )
            )
    )
    try {
        handle.returnValue.get()
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}

AggregatedMeteringCollectionFlow

Use the AggregatedMeteringCollectionFlow flow to collect aggregated metering data from other nodes on the network. You must specify the time window over which the flow will collect metering data, and the party running the node where metering data will be collected.

The flow returns the total count of signing events that happened on the monitored nodes during the time window.

You can use this flow to collect aggregated metering data from a remote node on the network. Aggregated metering data only contains the total number of signing events that happened within a given time period, without any additional information such as signing service public key, contract command, or transaction type.

The example below shows how to retrieve aggregated metering data by connecting to a node running on the local machine, from the node ran by O=PartyA,L=New York,C=US, for the duration of the past 7 days:

NetworkHostAndPort hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000");
CordaRPCClient client = new CordaRPCClient(hostAndPort);
AggregatedNodeMeteringData meteringData =
        client.use("rpcUsername", "rpcPassword", conn -> {
            CordaRPCOps rpcOps = conn.getProxy();
            Party destination = rpcOps.wellKnownPartyFromX500Name(
                    CordaX500Name.parse("O=PartyA,L=New York,C=US")
            );
            Instant now = Instant.now();
            Instant sevenDaysAgo = now.minus(7, ChronoUnit.DAYS);
            FlowHandle<AggregatedNodeMeteringData> handle =
                    rpcOps.startFlowDynamic(AggregatedMeteringCollectionFlow.class,
                            destination,
                            Filter.ByTimeStamp.between(sevenDaysAgo, now)
                    );
            Future<AggregatedNodeMeteringData> result = handle.getReturnValue();
            try {
                return result.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

val hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000")
val client = CordaRPCClient(hostAndPort)
val data = client.use("rpcUsername", "rpcPassword") { conn: CordaRPCConnection ->
    val rpcOps = conn.proxy
    val destination = rpcOps.wellKnownPartyFromX500Name(
            CordaX500Name.parse("O=PartyA,L=New York,C=US")
    )!!
    val now = Instant.now()
    val sevenDaysAgo = now.minus(7, ChronoUnit.DAYS)
    val handle = rpcOps.startFlow(
            ::AggregatedMeteringCollectionFlow,
            destination,
            Filter.ByTimeStamp.between(sevenDaysAgo, now)
    )
    try {
        handle.returnValue.get()
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}

MultiFilteredCollectionFlow

Use the MultiFilteredCollectionFlow flow to collect metering data from multiple nodes on the network by connecting to them via RPC. The MultiFilteredCollectionFlow flow is identical to the FilteredMeteringCollectionFlow flow except that MultiFilteredCollectionFlow collects data from multiple nodes on the network sequentially in a single flow, and returns the result as a JSON string.

You can use this flow only from the node shell. You can use the dedicated method FilteredMeteringCollectionFlow#multiCollect to collect metering data from multiple nodes in parallel by using an RPC client to connect to the initiating flow - the collection process then uses the flow framework to collect metering data from the other nodes on the network.

Instant now = Instant.now();
Instant sevenDaysAgo = now.minus(7, ChronoUnit.DAYS);
Filter filter = new Filter.And(
        Filter.ByTimeStamp.between(sevenDaysAgo, now),
        new Filter.Or(
                new Filter.ByCordapp.ByName("myCorDapp1"),
                new Filter.ByCordapp.ByName("myCorDapp2")
        )
);
CordaRPCOps rpcOps = conn.getProxy();
List<Pair<Party, Filter>> destinations =
        Stream.of("O=PartyA,L=New York,C=US", "O=PartyB,L=New York,C=US")
                .map(CordaX500Name::parse)
                .map(rpcOps::wellKnownPartyFromX500Name)
                .map(party -> new Pair<>(party, filter))
                .collect(Collectors.toList());

FilteredMeteringCollectionFlow.FilteredResultConsumer consumer = (
        Party destination,
        Filter f,
        Future<FilteredNodeMeteringData> result
) -> {
    try {
        FilteredNodeMeteringData data = result.get();
        //do something with the data
    } catch (ExecutionException | InterruptedException e) {
        throw new RuntimeException(e);
    }
};
FilteredMeteringCollectionFlow.multiCollect(
        rpcOps,
        destinations,
        consumer,
        Duration.of(30, ChronoUnit.SECONDS));

val now = Instant.now()
val sevenDaysAgo = now.minus(7, ChronoUnit.DAYS)
val filter = Filter.And(
        Filter.ByTimeStamp.between(sevenDaysAgo, now),
        Filter.Or(
                Filter.ByCordapp.ByName("myCorDapp1"),
                Filter.ByCordapp.ByName("myCorDapp2")
        )
)
val destinations = sequenceOf("O=PartyA,L=New York,C=US", "O=PartyB,L=New York,C=US")
        .map { CordaX500Name.parse(it) }
        .map { rpcOps.wellKnownPartyFromX500Name(it)!! }
        .map { it to filter }
        .toList()
val consumer = {
    destination: Party,
    filter : Filter,
    result: Future<FilteredNodeMeteringData> ->
    try {
        val data = result.get()
        //do something with the data
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}
FilteredMeteringCollectionFlow.multiCollect(
        rpcOps,
        destinations,
        consumer,
        Duration.of(30, ChronoUnit.SECONDS))

MultiAggregatedCollectionFlow

Use the MultiAggregatedCollectionFlow flow to collect aggregated metering data from multiple nodes on the network sequentially in a single flow. The MultiAggregatedCollectionFlow flow is identical to the AggregatedMeteringCollectionFlow flow except that MultiAggregatedCollectionFlow collects data from multiple nodes on the network sequentially in a single flow, and returns the result as a JSON string.

You can use this flow only from the node shell.

CordaRPCOps rpcOps = conn.getProxy();
Instant now = Instant.now();
Instant sevenDaysAgo = now.minus(7, ChronoUnit.DAYS);
List<Pair<Party, Filter>> destinations =
        Stream.of("O=PartyA,L=New York,C=US", "O=PartyB,L=New York,C=US")
                .map(CordaX500Name::parse)
                .map(rpcOps::wellKnownPartyFromX500Name)
                .map(party -> new Pair<>(party, Filter.ByTimeStamp.between(sevenDaysAgo, now)))
                .collect(Collectors.toList());

AggregatedMeteringCollectionFlow.AggregatedResultConsumer consumer = (
        Party destination,
        Filter f,
        Future<AggregatedNodeMeteringData> result
) -> {
    try {
        AggregatedNodeMeteringData data = result.get();
        //do something with the data
    } catch (ExecutionException | InterruptedException e) {
        throw new RuntimeException(e);
    }
};
AggregatedMeteringCollectionFlow.multiCollect(
        rpcOps,
        destinations,
        consumer,
        Duration.of(30, ChronoUnit.SECONDS));

val now = Instant.now()
val sevenDaysAgo = now.minus(7, ChronoUnit.DAYS)
val filter = Filter.ByTimeStamp.between(sevenDaysAgo, now)
val destinations = sequenceOf("O=PartyA,L=New York,C=US", "O=PartyB,L=New York,C=US")
        .map { CordaX500Name.parse(it) }
        .map { rpcOps.wellKnownPartyFromX500Name(it)!! }
        .map { it to filter }
        .toList()
val consumer = {
    _ : Party,
    _ : Filter,
    result: Future<AggregatedNodeMeteringData> ->
    try {
        val data = result.get()
        //do something with the data
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}
AggregatedMeteringCollectionFlow.multiCollect(
        rpcOps,
        destinations,
        consumer,
        Duration.of(30, ChronoUnit.SECONDS))

NotaryCollectionFlow

Use the NotaryCollectionFlow flow to collect metering data from notaries. You must specify the time window over which the flow will be collecting metering data. The flow output represents the total count of notarisation requests during the specified time window, along with a breakdown of these requests filtered by the parties that made them.

RetrieveCordappDataFlow

Use the RetrieveCordappDataFlow utility flow to extract CorDapp hashes and signing keys for a given CorDapp name for use in the NodeMeteringCollectionFlow flow filter. This flow provides information about the versions and vendors of the returned CorDapps so that the correct CorDapp data can be selected.

You can use this additional utility flow to retrieve CorDapp metadata for a particular CorDapp name, and to obtain CorDapp hashes and signing keys in the correct format in order to construct a valid MeteringFilter instance. The flow also returns the version numbers and vendors of the queried CorDapps, allowing you to extract the right hashes and keys for particular versions.

The example below shows how to invoke this flow from the node shell:

Tue May 05 16:23:08 IST 2020>>> flow start com.r3.corda.metering.RetrieveCordappDataFlow

 ✓ Starting
▶︎ Done
Flow completed with result: [{
  "name" : "Corda Finance Demo",
  "vendor" : "R3",
  "version" : "1",
  "hash" : "FD8C8A320794B7D928DB90EBACC27A06B6AB683111799380286F2F7A8AB819F2",
  "signingKeys" : [ "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" ]
}, {
  "name" : "Corda Finance Demo",
  "vendor" : "R3",
  "version" : "1",
  "hash" : "44489E8918D7D8F7A3227FE56EC34BFDDF15BD413FF92F23E72DD5D543BD6194",
  "signingKeys" : [ "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" ]
}, {
  "name" : "Corda Metering Collection Tool",
  "vendor" : "R3",
  "version" : "4.10-SNAPSHOT",
  "hash" : "FC0150EFAB3BBD715BDAA7F67B4C4DB5E133D919B6860A3D3B4C6C7D3EFE25D5",
  "signingKeys" : [ "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" ]
}]

The example below shows how to invoke this flow using RPC:

NetworkHostAndPort hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000");
CordaRPCClient client = new CordaRPCClient(hostAndPort);
List<? extends CordappData> corDappData =
  client.use("rpcUsername", "rpcPassword", conn -> {
    CordaRPCOps rpcOps = conn.getProxy();
    FlowHandle<List<? extends CordappData>> handle =
      rpcOps.startFlowDynamic(RetrieveCordappDataFlow.class);
    Future<List<? extends CordappData>> result = handle.getReturnValue();
    try {
        return result.get();
    } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
    }
});

val hostAndPort = NetworkHostAndPort.parse("127.0.0.1:10000")
val client = CordaRPCClient(hostAndPort)
val corDappData =
  client.use("rpcUsername", "rpcPassword") { conn: CordaRPCConnection ->
    val rpcOps = conn.proxy
    val handle = rpcOps.startFlow(::RetrieveCordappDataFlow)
    val result: Future<List<CordappData>> = handle.returnValue
    try {
        result.get()
    } catch (e: ExecutionException) {
        throw e.cause ?: e
    }
}

Collect metering data from multiple nodes

There are two mechanisms you can use to collect metering data from multiple nodes on the network - by connecting to the node via:

• the Corda RPC API
• the node shell

Collect data using the RPC API

You can use the following two methods:

• FilteredMeteringCollectionFlow.multiCollect
• AggregatedMeteringCollectionFlow.multicollect

Both methods start multiple parallel flows on the collector node, each of them collecting metering data from a different node on the network. You can specify a timeout period so that any flows that do not terminate within the timeout are simply cancelled, and only the data from the flows that completed successfully is processed.

You can specify a callback as an argument. The callback is invoked once for each destination node as soon as the relative flow returns. The callback takes the following parameters:

• The destination party from which metering data has been collected.
• The Filter instance that was used for the metering data collection.
• A Future that is guaranteed to be done at the time of the callback invocation.

If the flow invocation resulted in an exception, the exception is thrown again inside the callback when calling Future.get and it is expected that the callback is able to handle it. In case this fails, the execution is interrupted, and all the created sub-flows are cancelled.

Collect data using the node shell

You can use the following two flows:

• MultiAggregatedCollectionFlow
• MultiFilteredCollectionFlow

As regards data filtering and permissions, both of them are identical to their single node counterparts - AggregatedMeteringCollectionFlow and FilteredMeteringCollectionFlow, respectively.

They proceed through all the destination nodes in sequential order. This can significantly slow down the collection process and can cause the execution to hang indefinitely if one of the destination nodes is down or is not running the metering collection tool CorDapp.

Both flows take the following parameters when invoked from the shell:

• dateFormat: The date format to use for parsing the start and end parameter values. The accepted format is SimpleDateFormat.
• start: A string representing the start date to collect metering data from. The accepted format is either SimpleDateFormat or the one returned by the SimpleDateFormat.getInstance method, which defaults to your locale settings.
• end: A string representing the end date to collect metering data by. The accepted format is either SimpleDateFormat or the one returned by the SimpleDateFormat.getInstance method, which defaults to your locale settings.
• period: The period of time after start or before end to be used for collecting metering data. You can use the following measuring units: nanoseconds, microseconds, milliseconds, seconds, minutes, hours, days, weeks, months, and years. You can also use any unambiguous prefix for these units - for example, 1mo will be interpreted as one month but 1m will cause an error as it is unclear whether you mean one month or one minute. Any failure in the interpretation of this parameter will raise an exception IllegalArgumentException.
• destinations: A list of the X.500 names of the parties running the nodes from which metering data is to be collected. The names do not need to be the full qualified X.500 names since IdentityService.partiesFromName will be invoked on them and if more parties match input string, metering data will be collected from all their nodes. If this parameter is omitted, metering data will be collected from all nodes present in the network map.
• txTypes: A list of transaction types to be included in the results. If this parameter is omitted, transactions of any type will be collected. See the data filtering paragraph for more information about transaction types.
• filter: This parameter is only available for MultiFilteredCollectionFlow and allows you to filter metering data by CorDapp. See also the data filtering paragraph.

Output format

For both methods, the result printed in the shell terminal is a formatted JSON object. Its keys are the X.500 names of the destination nodes, and its value is the JSON representation of the response object returned from the collection - an instance of AggregatedNodeMeteringData for MultiAggregatedCollectionFlow, and an instance of FilteredNodeMeteringData for MultiFilteredCollectionFlow. If any of the destination nodes throws an exception, it is shown in the response object JSON.

Examples

Collecting aggregated metering data for last month from “O=PartyB,L=New York,C=US” and “O=PartyA,L=New York,C=US”

Wed May 06 15:13:52 IST 2020>>> flow start com.r3.corda.metering.MultiAggregatedCollectionFlow period: {value: 1mon}, destinations: [PartyA, PartyB]

▶︎ Starting
    Done
 ✓ Starting
▶︎ Done
Flow completed with result: {
  "data" : {
    "O=PartyB, L=New York, C=US" : {
      "version" : 1,
      "count" : 51
    },
    "O=PartyA, L=New York, C=US" : {
      "exception" : "net.corda.core.flows.FlowException",
      "message" : "com.r3.corda.metering.PermissionDeniedException: You don't have permission to collect aggregated metering from this node",
      "cause" : {
        "exception" : "com.r3.corda.metering.PermissionDeniedException",
        "message" : "You don't have permission to collect aggregated metering from this node"
      }
    }
  },
  "window" : {
    "startInstant" : "2020-04-06T14:14:38.844Z",
    "endInstant" : "2020-05-06T14:14:38.844Z"
  }
}

Alternatively, you can achieve a similar result by running the following commands:

flow start com.r3.corda.metering.MultiAggregatedCollectionFlow dateFormat: {value: "yyyy-MM-dd"},  period: {value: 1mon}, end : {value: "2020-05-06"}, destinations: [PartyA, PartyB]
flow start com.r3.corda.metering.MultiAggregatedCollectionFlow dateFormat: {value: "yyyy-MM-dd"},  start: {value: "2020-04-06"}, end : {value: "2020-05-06"}, destinations: [PartyA, PartyB]
flow start com.r3.corda.metering.MultiAggregatedCollectionFlow dateFormat: {value: "yyyy-MM-dd"},  start: {value: "2020-04-06"}, end : {value: "2020-05-06"}, destinations: [PartyA, PartyB]

Collecting filtered metering data for last month from O=PartyB,L=New York,C=US and "O=PartyA,L=New York,C=US"

Wed May 06 15:37:45 IST 2020>>> flow start com.r3.corda.metering.MultiFilteredCollectionFlow period: {value: 1mon}, destinations: [PartyA, PartyB], filter: {filterBy: CORDAPP_NAMES, values: [Finance]}, txTypes: [NORMAL]

 ✓ Starting
▶︎ Done
Flow completed with result: {
  "data" : {
    "O=PartyB, L=New York, C=US" : {
      "version" : 1,
      "entries" : [ {
        "signingId" : {
          "type" : "NODE_IDENTITY",
          "accountId" : null
        },
        "txType" : "NORMAL",
        "commands" : [ "net.corda.finance.contracts.asset.Cash.Commands.Issue" ],
        "count" : 1
      } ],
      "collectedCorDapps" : [ {
        "name" : "Corda Finance Demo",
        "vendor" : "R3",
        "version" : "1",
        "hash" : "4DF7DAC0703459E97CB040CD6194ACC0D7B53931FAFC859158B16FDD85D525B5",
        "signingKeys" : [ "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" ]
      } ]
    },
    "O=PartyA, L=New York, C=US" : {
      "version" : 1,
      "entries" : [ {
        "signingId" : {
          "type" : "NODE_IDENTITY",
          "accountId" : null
        },
        "txType" : "NORMAL",
        "commands" : [ "net.corda.finance.contracts.asset.Cash.Commands.Issue" ],
        "count" : 1
      } ],
      "collectedCorDapps" : [ {
        "name" : "Corda Finance Demo",
        "vendor" : "R3",
        "version" : "1",
        "hash" : "4DF7DAC0703459E97CB040CD6194ACC0D7B53931FAFC859158B16FDD85D525B5",
        "signingKeys" : [ "AA59D829F2CA8FDDF5ABEA40D815F937E3E54E572B65B93B5C216AE6594E7D6B" ]
      } ]
    }
  },
  "params" : {
    "window" : {
      "startInstant" : "2020-04-06T14:38:11.428Z",
      "endInstant" : "2020-05-06T14:38:11.428Z"
    },
    "filter" : {
      "filterBy" : "CORDAPP_NAMES",
      "values" : [ "Finance" ]
    }
  }
}

Alternatively, you can achieve a similar result by running the following commands:

flow start com.r3.corda.metering.MultiFilteredCollectionFlow period: {value: 1mon}, filter: {filterBy: CORDAPP_HASHES, values: [4DF7DAC0703459E97CB040CD6194ACC0D7B53931FAFC859158B16FDD85D525B5]}
flow start com.r3.corda.metering.MultiFilteredCollectionFlow dateFormat: {value: "yyyy-MM-dd"},  start: {value: "2020-04-06"}, end : {value: "2020-05-06"}, destinations: [PartyA, PartyB], filter: {filterBy: CORDAPP_NAMES, values: [Finance]}, txTypes: [NORMAL]
flow start com.r3.corda.metering.MultiFilteredCollectionFlow dateFormat: {value: "yyyy-MM-dd"},  start: {value: "2020-04-06"}, end : {value: "2020-05-06"}, destinations: [PartyA, PartyB], filter: {filterBy: CORDAPP_NAMES, values: [Finance]}, txTypes: [NORMAL]

Filter data using the node shell

When you use the node shell to filter metering data, you can only filter by:

• CorDapp
• transaction type
• timestamp

Filter by CorDapps

To filter metering data by CorDapp, use the filter parameter in the MeteringCollectionFlow, MultiAggregatedMeteringCollectionFlow, and MultiFilteredMeteringCollectionFlow flows.

This parameter requires an object created by the filterBy parameter that specifies the type of filter and the filter values - an array of strings that represents the filter argument. The following example shows a list of filters available for filterBy:

Filter by transaction type

To filter metering data by transaction type, use the txTypes parameter in the MultiAggregatedMeteringCollectionFlow and MultiFilteredMeteringCollectionFlow flows.

This parameter takes an array with all the types that will be included.

The available transaction types are as follows:

• NORMAL
• CONTRACT_UPGRADE
• NOTARY_CHANGE
• UNKNOWN

Filter data using the RPC API

You can use data filtering via the RPC API for the NodeMeteringCollectionFlow, AggregatedMeteringCollectionFlow, and FilteredMeteringCollectionFlow flows.

All classes listed below belong to the com.r3.corda.metering.filter package.