feature: Show partition size

[sehz]

in the CLI, show the size of the data for each partition:

$ fluvio partition list
 TOPIC      PARTITION  LEADER  REPLICAS  RESOLUTION  SIZE HW     LEO    LRS  FOLLOWER OFFSETS 
 longevity      0       5001    []        Online      4G  88455  88455  0    []       

In order to do this, we need to add size to the status reported by SPU to SC

[Afourcat]

Hello,
I was trying to solve this issue, but, I feel a little bit confused, could someone tell me if I am on the right track and/or maybe give me a hint?

First of all, is the "size" the space taken, in bytes, by the stored records in the SPU?

CLI Side

On the CLI, data shown in the fluvio partition list output are fetched from here:

fluvio/crates/fluvio-cli/src/partition/list.rs

Line 32 in 0b99188

let partitions = admin.list::<PartitionSpec, _>(vec![]).await?;

In the metadata I need to add:

Metadata {
    name: "cat-facts-3-partionned-4",
    spec: PartitionSpec {
        leader: 0,
        replicas: [
            0,
        ],
        cleanup_policy: None,
        storage: None
    },
    status: PartitionStatus {
        resolution: Online,
        leader: ReplicaStatus {
            spu: 0,
            hw: 5,
            leo: 5,
        },
        lsr: 0,
        size: 240000, // size of the partition in bytes ?
        replicas: [],
        is_being_deleted: false,
    },
},

SPU Side

If I understand well enough, the following function sends back the status from the SPU's Leader to the SC.

fluvio/crates/fluvio-spu/src/control_plane/dispatcher.rs

Lines 189 to 208 in 0b99188

	/// send status back to sc, if there is error return false
	#[instrument(skip(self))]
	async fn send_status_back_to_sc(
	&mut self,
	sc_sink: &mut FluvioSink,
	) -> Result<(), SocketError> {
	let requests = self.status_update.remove_all().await;
	if requests.is_empty() {
	trace!("sending empty status");
	} else {
	trace!(requests = ?requests, "sending status back to sc");
	}
	let message = RequestMessage::new_request(UpdateLrsRequest::new(requests));
	sc_sink.send_request(&message).await.map_err(|err| {
	error!("error sending status back: {:#?}", err);
	err
	})
	}

It seems that I have to add a field somewhere, in this UpdateLrsRequest, but where exactly and how to fetch partitions' size is unclear.

If I'm right, partitions can be spread across multiple SPUs. So, do I need to gather partitions' size from followers and leaders, and send them to the sc through the UpdateLrsRequest?

Or am I completely wrong?

Thank you for your time,

[sehz]

Yes, this requires adding sizing API to ReplicaStorage in: https://github.com/infinyon/fluvio/blob/master/crates/fluvio-storage/src/lib.rs.

[Afourcat]

While implementing the feature, I met a compilation error.
The fluvio-future crate is used in the fluvio-storage crate with the feature fs.
Which triggers a build error because the fs feature doesn't pull the optional async-trait crate.

fs = ["async-fs", "futures-lite", "pin-utils"]

Error:

$ cargo build --features "fs"
error: cannot determine resolution for the attribute macro `async_trait`
  --> src/fs/extension.rs:34:3
   |
34 | #[async_trait]
   |   ^^^^^^^^^^^
   |
   = note: import resolution is stuck, try simplifying macro imports

This could be fix in two ways:

• add the async-trait dependency to the fs feature in the fluvio-future/Cargo.toml file.
• add the async-trait dependency to the fluvio-future dependency in the flavio-storage/Cargo.toml file.

#                                                                             vvvvvvvvvvv
fluvio-future = { version = "0.3.13", features = ["fs", "mmap", "zero_copy", "async-trait"] }

I think that the first one, is the right choice, but, to help myself focus on the current issue, I implemented the second one, which let me test right away.

Let me know if you want me to fill an issue on the first,

Thank you,

[sehz]

You can build cluster:

make build-cluster

They are described in: https://github.com/infinyon/fluvio/blob/master/DEVELOPER.md

[Afourcat]

I finally have a working version,

$ flvd partition list
TOPIC            PARTITION  LEADER  REPLICAS  RESOLUTION  SIZE   HW   LEO  LRS  FOLLOWER OFFSETS 
 cat-facts        0          5001    []        Online     31236  139  139  0    [] 
 flex-facts       0          5001    []        Online     1072   3    3    0    [] 
 flex-facts       1          5001    []        Online     4310   3    3    0    [] 
 flex-facts       2          5001    []        Online     4310   3    3    0    [] 
 flex-facts       3          5001    []        Online     159    2    2    0    [] 
 flex-facts       4          5001    []        Online     680    2    2    0    [] 
 multi-spu-facts  0          5001    [5002]    Online     4876   10   10   1    [ReplicaStatus { spu: 5002, hw: 10, leo: 10 }] 
 multi-spu-facts  1          5002    [5001]    Online     8329   10   10   1    [ReplicaStatus { spu: 5001, hw: 10, leo: 10 }] 
 multi-spu-facts  2          5001    [5002]    Online     10509  10   10   1    [ReplicaStatus { spu: 5002, hw: 10, leo: 10 }] 
 multi-spu-facts  3          5002    [5001]    Online     7535   10   10   1    [ReplicaStatus { spu: 5001, hw: 10, leo: 10 }] 
 multi-spu-facts  4          5001    [5002]    Online     7831   10   10   1    [ReplicaStatus { spu: 5002, hw: 10, leo: 10 }]

As you can see I have 3 topics:

• cat-facts which has 1 replica and 1 partition
• flex-facts which has 1 replica and 5 partitions
• multi-spu-facts which has 2 replicas and 5 partitions

If I print the precise size of logs:

$ #  cat-facts        0          5001    []        Online     31236  139  139  0    [] 
$ stat -c '%n %s' ./log/spu-logs-5001/cat-facts-0/00000000000000000000.log 
./log/spu-logs-5001/cat-facts-0/00000000000000000000.log 31236

$ # flex-facts       0          5001    []        Online     1072   3    3    0    [] 
$ stat -c '%n %s' ./log/spu-logs-5001/flex-facts-0/00000000000000000000.log
./log/spu-logs-5001/flex-facts-0/00000000000000000000.log 1072

$ #  flex-facts       3          5001    []        Online     159    2    2    0    [] 
$ stat -c '%n %s' ./log/spu-logs-5001/flex-facts-3/00000000000000000000.log
./log/spu-logs-5001/flex-facts-3/00000000000000000000.log 159

Right now, when a partition is spread on multiple SPUs I only have the size of the leader.

$ #  multi-spu-facts  0          5001    [5002]    Online     4876   10   10   1    [ReplicaStatus { spu: 5002, hw: 10, leo: 10 }] 
$ # SPU 5001
$ stat -c '%n %s' ./log/spu-logs-5001/multi-spu-facts-0/00000000000000000000.log 
./log/spu-logs-5001/multi-spu-facts-0/00000000000000000000.log 4876
# SPU 5002
$ stat -c '%n %s' ./log/spu-logs-5002/multi-spu-facts-0/00000000000000000000.log
./log/spu-logs-5002/multi-spu-facts-0/00000000000000000000.log 4876

Size is correctly updated when the replica receives records and writes them to the log file.

Next steps

• Print size with human-readable string: 10K, 32M, 4G, 3T?
• Print the total size of the partition over all replicas:
  • Simply multiply: (leader size * replicas number)?
  • Forward from followers updates?
• Proper error handling
• Unit tests

[sehz]

Wow, nice!. A human-readable size would be nice! (See fluvio-cluster's diagnostic).
For now, the size for partition leader is sufficient.

[Afourcat]

Hello again,
I have a question about the actual implementation.

Here is the function that creates the LrsRequest that will ultimately be sent to the SC.

async fn as_lrs_request(&self) -> LrsRequest {
    let leader = (self.leader(), self.hw(), self.leo()).into();
    let replicas: Vec<ReplicaStatus> = self
        .followers
        .read()
        .await
        .iter()
        .map(|(follower_id, follower_info)| {
            (*follower_id, follower_info.hw, follower_info.leo).into()
        })
        .collect();
    let storage_reader = self.storage.read().await;
    let size = storage_reader
        .get_partition_size()
        .await
        .expect("Unexpected error") as i64; // TODO: Change
                                                                       
    LrsRequest::new(self.id().to_owned(), leader, replicas, size)
}

When fetching partition size

let storage_reader = self.storage.read().await;
let size = storage_reader
    .get_partition_size()
    .await
    .expect("Unexpected error") as i64; // TODO: Change

What kind of error are we expecting in order to replace the expect?

• Should we send size of -1?
• Should we propagate an error and skip the the LrsRequest packet?
  Or something else?

Also,
From your prior response,

A human-readable size would be nice! (See fluvio-cluster's diagnostic)...

I checked the crate crates/fluvio-cluster, went to the diagnostic file, read though it, and then I ran a diagnostic from the CLI.
But, I think I'm missing something here considering that I could not find anything helpful about human-readable storage size.

However after some research on the internet, I found this package.
Which may become useful,

Thank you,

[sehz]

human_byte seems reasonable. For error, if there is an error, just create display representation of error instead of size.