Commit 3f307b14 authored by Geoff Simmons's avatar Geoff Simmons

Refactor querying for addresses of nodes in a self-sharding cluster.

This had been done by querying Pods. It's enough just to query
Endpoints.
parent ca3b99d8
...@@ -277,68 +277,53 @@ func (worker *NamespaceWorker) configSharding(spec *vcl.Spec, ...@@ -277,68 +277,53 @@ func (worker *NamespaceWorker) configSharding(spec *vcl.Spec,
svc.Namespace, svc.Name) svc.Namespace, svc.Name)
endps, err := worker.getAllServiceEndpoints(svc) endps, err := worker.getAllServiceEndpoints(svc)
if endps == nil || len(endps) == 0 || err != nil { if err != nil {
return IncompleteIfNotFound(
err, "Error getting endpoints for service %s/%s: %v",
svc.Namespace, svc.Name, err)
}
if endps == nil || len(endps) == 0 { if endps == nil || len(endps) == 0 {
return update.MakeIncomplete( return update.MakeIncomplete(
"could not find endpoints for service: %s/%s", "could not find endpoints for service: %s/%s",
svc.Namespace, svc.Name) svc.Namespace, svc.Name)
} }
return IncompleteIfNotFound(
err, "Error getting endpoints for service %s/%s: %v",
svc.Namespace, svc.Name, err)
}
worker.log.Tracef("Endpoints for shard configuration: %+v", endps) worker.log.Tracef("Endpoints for shard configuration: %+v", endps)
var nAddrs int // Populate spec.ShardCluster.Nodes with Pod names and the http endpoint
var httpPort int32
for _, eps := range endps { for _, eps := range endps {
for _, endpSubset := range eps.Subsets { for _, endpSubset := range eps.Subsets {
nAddrs += len(endpSubset.Addresses) httpPort := int32(0)
nAddrs += len(endpSubset.NotReadyAddresses)
for _, port := range endpSubset.Ports { for _, port := range endpSubset.Ports {
if httpPort == 0 && port.Name == "http" { if port.Name == "http" {
httpPort = port.Port httpPort = port.Port
} break
}
} }
} }
if httpPort == 0 { if httpPort == 0 {
return update.MakeFatal( continue
"No http port found in the endpoints for service %s/%s",
svc.Namespace, svc.Name)
} }
if nAddrs <= 1 { addresses := append(endpSubset.Addresses,
return update.MakeFatal( endpSubset.NotReadyAddresses...)
"Sharding requested, but %d endpoint addresses found "+ for _, addr := range addresses {
"for service %s/%s", nAddrs, svc.Namespace, node := vcl.Service{
svc.Name) Addresses: make([]vcl.Address, 1),
} }
ns, name := getTargetPod(addr)
pods, err := worker.getPods(svc) if ns != "" && name != "" {
if err != nil { node.Name = ns + "_" + name
return IncompleteIfNotFound(err,
"Error getting pod information for service %s/%s: %v",
svc.Namespace, svc.Name, err)
} }
if len(pods.Items) <= 1 { node.Addresses[0].IP = addr.IP
return update.MakeFatal( node.Addresses[0].Port = httpPort
"Sharding requested, but %d pods found for service "+ spec.ShardCluster.Nodes =
"%s/%s", len(pods.Items), svc.Namespace, append(spec.ShardCluster.Nodes, node)
svc.Name)
} }
worker.log.Tracef("Pods for shard configuration: %+v", pods.Items)
// Populate spec.ShardCluster.Nodes with Pod names and the http endpoint
for _, pod := range pods.Items {
node := vcl.Service{Addresses: make([]vcl.Address, 1)}
if pod.Spec.Hostname != "" {
node.Name = pod.Spec.Hostname
} else {
node.Name = pod.Name
} }
node.Addresses[0].IP = pod.Status.PodIP }
node.Addresses[0].Port = httpPort if len(spec.ShardCluster.Nodes) <= 1 {
spec.ShardCluster.Nodes = append(spec.ShardCluster.Nodes, node) return update.MakeFatal(
"Sharding requested, but %d endpoint addresses found "+
"for service %s/%s",
len(spec.ShardCluster.Nodes), svc.Namespace, svc.Name)
} }
worker.log.Tracef("Node configuration for self-sharding in Service "+ worker.log.Tracef("Node configuration for self-sharding in Service "+
"%s/%s: %+v", svc.Namespace, svc.Name, spec.ShardCluster.Nodes) "%s/%s: %+v", svc.Namespace, svc.Name, spec.ShardCluster.Nodes)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment