Ozone ofs in presto

[lukmanulhakkeem]

Hi folks, I am able to connect to ozone using the iceberg catalog from presto using the s3 gateway. But is it possible to connect to ozone from presto using the ofs protocol? Did anyone try this?

[ivandika3]

Yes, ofs is a Hadoop compatible protocol and should be able to be used by Presto. Our Ozone cluster is currently being used by Presto to query Hive tables on Ozone (instead of HDFS).

[lukmanulhakkeem]

Hi @ivandika3 Thanks for the quick response. I tried the OFS protocol by adding the jar hadoop-ozone-filesystem-hadoop3 in presto. From Presto, I am able to create the schema and table(confirmed the directories from the ozone) but I am getting the below error when I am inserting data into the table from Presto. I am trying to connect with a docker version of apache ozone with 3 datanodes. It seems like the client(ozone client in presto) is getting the internal IP of the datanode inside the docker container hence it is not accessible. In presto I am only providing the om host as the input.

2024-10-03T16:41:54.022+0530 WARN 20241003_105317_00001_xiv52.1.0.0.0-16-142 org.apache.hadoop.ozone.client.io.KeyOutputStream Encountered exception java.io.IOException: Unexpected Storage Container Exception: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.ratis.protocol.exceptions.AlreadyClosedException: SlidingWindow$Client:client-C13B7EE9847D->RAFT is closed. on the pipeline Pipeline[ Id: 1b11d147-c13e-45f9-9b08-c15280dc93ee, Nodes: e60e96aa-9646-4bde-8f7a-89ea99f61dfd{ip: 172.18.0.8, host: root_datanode_3.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}2b16f663-d0dd-4ef2-ae8b-fb2346f152ff{ip: 172.18.0.7, host: root_datanode_2.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}0652c188-04ed-478d-a6d2-3493e64ceb5f{ip: 172.18.0.6, host: root_datanode_1.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}, Type:RATIS, Factor:THREE, State:OPEN, leaderId:e60e96aa-9646-4bde-8f7a-89ea99f61dfd, CreationTimestamp2024-09-20T09:46:50.438Z]. The last committed block length is 0, uncommitted data length is 411 retry count 0

[ivandika3]

@lukmanulhakkeem Does the client runs on the docker environment as well?

Maybe you can take a look at @adoroszlai patch #5016 for docker-compose setup and script for Trino, Iceberg, and Ozone. I'm personally not familiar with the details for Trino/Presto on Ozone, just saying that it's possible to run Trino/Presto on top of Ozone.

Edit: Seems #5016 runs it on top of S3G, instead of ofs.

[lukmanulhakkeem]

No, the client is running in my local. As you mentioned the above PR is using S3G.

Is there any idea how this internal IP is getting to the client. I am not aware of the internal communication inside the ozone. I guess it will work If I managed to supply the public IP instead of internal IP to the client. But not sure how it can be done.

[ivandika3]

Yes, maybe because the datanode IPs was not exposed to localhost. Can try upload the key using ozone cli using ozone sh key put or ofs -put command from local. Or maybe you can try to run the presto client from one of the docker containers in Ozone (e.g. OM)

I'll try to come up with PoC when I have time.

[lukmanulhakkeem]

Sure @ivandika3 i will try it out.
Can I deploy ozone in a single machine without docker?

[ivandika3]

Yes, usually I will follow https://cwiki.apache.org/confluence/display/OZONE/Run+Ozone+cluster+from+IDE to setup local Ozone cluster from Intellij.

Can try this order:

• StorageContainerManagerInit (to initialize the SCM dir)
• StorageContainerManger (to start SCM)
• OzoneManagerInit (to initialize OM, it requires SCM)
• OzoneManager
• Datanode1, Datanode2, Datanode3

Afterwards, you can do some sanity checking first by uploading 1 or 2 objects.

[lukmanulhakkeem]

This is not only the issue from presto. I followed the below documentation to try locally using hadoop
https://ozone.apache.org/docs/1.0.0/interface/ofs.html#:~:text=at%20the%20moment.-,Configuration,-Please%20add%20the

I can create volume and schema using hdfs dfs -mkdir /volume1 and hdfs dfs -mkdir /volume1/bucket1
But getting the same error when I am performing put operation hdfs dfs -put /etc/hosts /volume1/bucket1/test. Posting the error below

2024-10-10 11:32:27,878 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2024-10-10 11:32:33,082 INFO impl.MetricsConfig: Loaded properties from hadoop-metrics2.properties
2024-10-10 11:32:38,150 INFO impl.MetricsSystemImpl: Scheduled Metric snapshot period at 10 second(s).
2024-10-10 11:32:38,151 INFO impl.MetricsSystemImpl: XceiverClientMetrics metrics system started
2024-10-10 11:32:38,423 INFO metrics.MetricRegistries: Loaded MetricRegistries class org.apache.ratis.metrics.impl.MetricRegistriesImpl
2024-10-10 11:32:38,424 INFO metrics.RatisMetrics: Creating Metrics Registry : ratis.client_message_metrics.client-D1C8B439F669->3a3cacdc-72d3-408b-84f9-a5b853781093
2024-10-10 11:32:38,425 WARN impl.MetricRegistriesImpl: First MetricRegistry has been created without registering reporters. You may need to call MetricRegistries.global().addReporterRegistration(...) before.
2024-10-10 11:33:11,952 WARN grpc.GrpcUtil: Timed out gracefully shutting down connection: ManagedChannelOrphanWrapper{delegate=ManagedChannelImpl{logId=1, target=172.18.0.7:9858}}. 
2024-10-10 11:33:19,142 INFO metrics.RatisMetrics: Creating Metrics Registry : ratis.client_message_metrics.client-D1C8B439F669->96ca632a-1667-415d-ba94-bf4b1f9fb958
2024-10-10 11:33:19,142 WARN impl.MetricRegistriesImpl: First MetricRegistry has been created without registering reporters. You may need to call MetricRegistries.global().addReporterRegistration(...) before.
2024-10-10 11:33:52,176 WARN grpc.GrpcUtil: Timed out gracefully shutting down connection: ManagedChannelOrphanWrapper{delegate=ManagedChannelImpl{logId=8, target=172.18.0.5:9858}}. 
2024-10-10 11:34:27,721 WARN grpc.GrpcUtil: Timed out gracefully shutting down connection: ManagedChannelOrphanWrapper{delegate=ManagedChannelImpl{logId=15, target=172.18.0.7:9858}}. 
2024-10-10 11:35:06,348 WARN grpc.GrpcUtil: Timed out gracefully shutting down connection: ManagedChannelOrphanWrapper{delegate=ManagedChannelImpl{logId=22, target=172.18.0.5:9858}}. 
2024-10-10 11:35:41,453 ERROR storage.BlockOutputStream: writing chunk failed 113750153625600007_chunk_1 blockID conID: 2 locID: 113750153625600007 bcsId: 0 with exception org.apache.ratis.protocol.exceptions.AlreadyClosedException: SlidingWindow$Client:client-D1C8B439F669->RAFT is closed.
2024-10-10 11:35:41,468 WARN io.KeyOutputStream: Encountered exception java.io.IOException: Unexpected Storage Container Exception: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.ratis.protocol.exceptions.AlreadyClosedException: SlidingWindow$Client:client-D1C8B439F669->RAFT is closed. on the pipeline Pipeline[ Id: f0aa8815-775d-41e4-a2ce-4ae5d2313c26, Nodes: c724aaf9-4827-4e3c-8601-99b5f4ab3bb7{ip: 172.18.0.8, host: root_datanode_2.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}3a3cacdc-72d3-408b-84f9-a5b853781093{ip: 172.18.0.7, host: root_datanode_3.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}96ca632a-1667-415d-ba94-bf4b1f9fb958{ip: 172.18.0.5, host: root_datanode_1.root_default, ports: [REPLICATION=9886, RATIS=9858, RATIS_ADMIN=9857, RATIS_SERVER=9856, STANDALONE=9859], networkLocation: /default-rack, certSerialId: null, persistedOpState: IN_SERVICE, persistedOpStateExpiryEpochSec: 0}, Type:RATIS, Factor:THREE, State:OPEN, leaderId:3a3cacdc-72d3-408b-84f9-a5b853781093, CreationTimestamp2024-10-10T05:27:20.006Z]. The last committed block length is 0, uncommitted data length is 412 retry count 0

[lukmanulhakkeem]

@ivandika3 @adoroszlai - So presto should be there in the same machine where ozone om is installed right? I am assuming this would be the same regardless of my setup like docker or in a physical cluster. Please correct me If I am wrong

[ivandika3]

@lukmanulhakkeem Here's the client requirements:

• Client is able to connect to OM (default port 9862): client can do metadata operations (e.g. create volume / create bucket)
• Client is able to connect to DN (default port 19864): client can upload and download file

So not necessarily the same machine, you can whether you can connect by using ping or telnet on OM and DN.

[ivandika3]

I am assuming this would be the same regardless of my setup like docker or in a physical cluster.

No, docker will create an internal network that cannot be accessed from outside unless you use port forwarding. Even if you port forward the datanodes, the datanode network address returned by SCM and OM to client will be the docker internal network address, which cannot be used by client to connect from outside docker network.

[lukmanulhakkeem]

Yes I understand that. But will that be the same if I am having ozone in a physical machine, for eg say ozone in Cloudera.

[ivandika3]

Not familiar with Ozone on Cloudera, but should be similar client setup.

[lukmanulhakkeem]

Does the client directly interact with data nodes both during read and write?

[ivandika3]

Yes, client will directly connect to datanodes for read and write ops.

[lukmanulhakkeem]

@ivandika3 - I tried presto and ozone in the same machine. This is working with hive catalog in presto.

But I tried the remote connectivity to an ozone instance in cloudera. It throws the same error I posted in the previous thread. The client is trying to access the internal IP of the datanode which is in the cloudera VM. So this is not only with the docker setup.

Is there a way I can specify datanode host at the client side using ozone-site.xml or hdfs-site.xml?

[ivandika3]

Not sure how Cloudera VM is setup, maybe the other Cloudera community members can help (cc: @ChenSammi). However, as said earlier, I don't think the client can access datanode internal IPs.

Is there a way I can specify datanode host at the client side using ozone-site.xml or hdfs-site.xml?

No such configuration is supported currently. Since datanodes can be added dynamically, client configuration for the accessible IP of the datanodes need to be updated every time there is a new datanode added, making it infeasible.

There might be some solutions that can be implemented, such as a client-side configuration to automatically convert a certain internal address to address accessible to the client.

[lukmanulhakkeem]

Yes, I am trying for the IP forward.
If possible could you specify where we are using the datanode host address in the code? . So that I can create a config to read the datanode host.

[ivandika3]

For read client, see XceiverClientGrpc#connectToDatanode. For write client, see XceiverClientRatis#connect and RatisHelper#toRaftPeerAddress. Both will use DatanodeDetails#getIpAddress.

RatisHelper#toRaftPeerAddress seems to have an option to access host name by setting dfs.datanode.use.datanode.hostname, but not sure if it's well tested.

[lukmanulhakkeem]

Thanks @ivandika3 . I tried the property dfs.datanode.use.datanode.hostname. It was not working for me. I will check once again. Above details are enough for the changes. Thanks.

[adoroszlai]

@lukmanulhakkeem try using Docker's host network driver. This eliminates container-specific IP addresses and NAT. On the flip side, you'll need to configure different set of ports for each service with multiple instances (e.g. each datanode) to avoid port conflict.