Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

🌱 test/inmemory: use port only to identify the wcl to make port-forward… #10245

Merged
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 23 additions & 14 deletions test/infrastructure/inmemory/pkg/server/mux.go
Original file line number Diff line number Diff line change
Expand Up @@ -108,8 +108,8 @@ type WorkloadClustersMux struct {
debugServer http.Server
muxServer http.Server
workloadClusterListeners map[string]*WorkloadClusterListener
// workloadClusterNameByHost maps from Host to workload cluster name.
workloadClusterNameByHost map[string]string
// workloadClusterNameByPort maps from Port to workload cluster name.
workloadClusterNameByPort map[string]string

lock sync.RWMutex
log logr.Logger
Expand All @@ -131,7 +131,7 @@ func NewWorkloadClustersMux(manager inmemoryruntime.Manager, host string, opts .
portIndex: options.MinPort,
manager: manager,
workloadClusterListeners: map[string]*WorkloadClusterListener{},
workloadClusterNameByHost: map[string]string{},
workloadClusterNameByPort: map[string]string{},
log: log.Log,
}

Expand Down Expand Up @@ -169,7 +169,12 @@ func (m *WorkloadClustersMux) mixedHandler() http.Handler {
resourceGroupResolver := func(host string) (string, error) {
m.lock.RLock()
defer m.lock.RUnlock()
wclName, ok := m.workloadClusterNameByHost[host]

_, port, err := net.SplitHostPort(host)
if err != nil {
return "", err
}
wclName, ok := m.workloadClusterNameByPort[port]
if !ok {
return "", errors.Errorf("failed to get workloadClusterListener for host %s", host)
}
Expand Down Expand Up @@ -211,18 +216,22 @@ func (m *WorkloadClustersMux) getCertificate(info *tls.ClientHelloInfo) (*tls.Ce
defer m.lock.RUnlock()

// Identify which workloadCluster/resourceGroup a request targets to.
hostPort := info.Conn.LocalAddr().String()
wclName, ok := m.workloadClusterNameByHost[hostPort]
_, port, err := net.SplitHostPort(info.Conn.LocalAddr().String())
if err != nil {
return nil, err
}

wclName, ok := m.workloadClusterNameByPort[port]
if !ok {
err := errors.Errorf("failed to get listener name for workload cluster serving on %s", hostPort)
err := errors.Errorf("failed to get listener name for workload cluster serving on %s", port)
m.log.Error(err, "Error resolving certificates")
return nil, err
}

// Gets the listener config for the target workloadCluster.
wcl, ok := m.workloadClusterListeners[wclName]
if !ok {
err := errors.Errorf("failed to get listener with name %s for workload cluster serving on %s", wclName, hostPort)
err := errors.Errorf("failed to get listener with name %s for workload cluster serving on %s", wclName, port)
m.log.Error(err, "Error resolving certificates")
return nil, err
}
Expand All @@ -231,12 +240,12 @@ func (m *WorkloadClustersMux) getCertificate(info *tls.ClientHelloInfo) (*tls.Ce
// NOTE: the port forward call to etcd sets the server name to the name of the targeted etcd pod,
// which is also the name of the corresponding etcd member.
if wcl.etcdMembers.Has(info.ServerName) {
m.log.V(4).Info("Using etcd serving certificate", "listenerName", wcl, "host", hostPort, "etcdPod", info.ServerName)
m.log.V(4).Info("Using etcd serving certificate", "listenerName", wcl, "host", port, "etcdPod", info.ServerName)
return wcl.etcdServingCertificates[info.ServerName], nil
}

// Otherwise we assume the request targets the API server.
m.log.V(4).Info("Using API server serving certificate", "listenerName", wcl, "host", hostPort)
m.log.V(4).Info("Using API server serving certificate", "listenerName", wcl, "host", port)
return wcl.apiServerServingCertificate, nil
}

Expand Down Expand Up @@ -320,7 +329,7 @@ func (m *WorkloadClustersMux) initWorkloadClusterListenerWithPortLocked(wclName
// NOTE: it is required to add on both maps and keep them in sync
// In order to get the resourceGroupResolver to work.
m.workloadClusterListeners[wclName] = wcl
m.workloadClusterNameByHost[wcl.HostPort()] = wclName
m.workloadClusterNameByPort[fmt.Sprintf("%d", wcl.Port())] = wclName

m.log.Info("Workload cluster listener created", "listenerName", wclName, "address", wcl.Address())
return wcl
Expand Down Expand Up @@ -432,9 +441,9 @@ func (m *WorkloadClustersMux) AddAPIServer(wclName, podName string, caCert *x509
return nil
}

l, err := net.Listen("tcp", wcl.HostPort())
l, err := net.Listen("tcp", fmt.Sprintf(":%d", wcl.Port()))
if err != nil {
return errors.Wrapf(err, "failed to start WorkloadClusterListener %s, %s", wclName, wcl.HostPort())
return errors.Wrapf(err, "failed to start WorkloadClusterListener %s, %s", wclName, fmt.Sprintf(":%d", wcl.Port()))
}
wcl.listener = l

Expand Down Expand Up @@ -603,7 +612,7 @@ func (m *WorkloadClustersMux) DeleteWorkloadClusterListener(wclName string) erro
}

delete(m.workloadClusterListeners, wclName)
delete(m.workloadClusterNameByHost, wcl.HostPort())
delete(m.workloadClusterNameByPort, fmt.Sprintf("%d", wcl.Port()))

m.log.Info("Workload cluster listener deleted", "listenerName", wclName, "address", wcl.Address())
return nil
Expand Down
Loading