Add dataproxy client selector

[katrogan]

NOT TO BE MERGED UNTIL BE CHANGES ARE IN.

This PR adds a new dependency on the flyte remote client for dataproxy endpoints

1. Use a client conn cache for dataproxy operations, using 'operation' and 'resource' as cache keys
2. On cache misses call SelectCluster to get the respective cluster endpoint and initialize and cache a new client conn

Going forward all new dataproxy calls should use this pattern.

Today this PR updates the only uses of dataproxy in the sdk: UploadInputs and CreateUploadLocation

---

Testing

Verified that the CreateRun path (which calls UploadInputs now) works with a local cluster

flyte -vvv  --config .flyte/config-oss-local.yaml  run -p flytesnacks -d development examples/basics/hello.py main
...
╭────────────────────────────────────────── Remote Run ───────────────────────────────────────────╮
│ Created Run: rmj7wwft78l69stlj69j                                                               │
│ URL: http://localhost:30080/v2/domain/development/project/flytesnacks/runs/rmj7wwft78l69stlj69j │
╰─────────────────────────────────────────────────────────────────────────────────────────────────╯

ref 26-353

[EngHabu]

LGTM, @pingsutw @wild-endeavor should advise though

[EngHabu]

Should this operation be an enum instead?

[katrogan]

done

[EngHabu]

Is this the best way to create the one of 😬, @wild-endeavor @pingsutw ?

[pingsutw]

we can do something like

  req = SelectClusterRequest(operation=operation)                  
  if hasattr(resource, "DESCRIPTOR"):                                                                    
      oneof = req.DESCRIPTOR.oneofs_by_name["resource"]  # replace with actual oneof name
      for field in oneof.fields:                                                                         
          if field.message_type is resource.DESCRIPTOR:                                                  
              getattr(req, field.name).CopyFrom(resource)                                                
              break    

[katrogan]

nice thanks. done

[EngHabu]

Are you going to normalize this here? stripping/adding http/s or dns:/// ?

[katrogan]

not necessary! create_session_config already calls normalize_rpc_endpoint

[EngHabu]

Let's make sure we are throwing informative errors here in the infamous "no healthy clusters" error.. we now have a good place to catch that and raise a good error to the user

[katrogan]

updated