github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-09-08 04:32:37 +00:00
Code Issues Projects Releases Wiki Activity

Migrated ipvs/proxier.go(remaining) to structured logging (#105770)

Browse Source 
* fixed improperly migrated logs

* fixed improperly migrated logs

* add
This commit is contained in:
Shivanshu Raj Shrivastava
2021-10-21 21:11:57 +05:30
committed by GitHub
parent 4cfd2820ba
commit 86aed49a18
1 changed files with 26 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
pkg/proxy/ipvs/proxier.go
View File

@@ -1398,10 +1398,10 @@ func (proxier *Proxier) syncProxyRules() {
activeIPVSServices[serv.String()] = true activeIPVSServices[serv.String()] = true
activeBindAddrs[serv.Address.String()] = true activeBindAddrs[serv.Address.String()] = true
if err := proxier.syncEndpoint(svcName, svcInfo.NodeLocalExternal(), serv); err != nil { if err := proxier.syncEndpoint(svcName, svcInfo.NodeLocalExternal(), serv); err != nil {
klog.ErrorS(err, "Failed to sync endpoint for service", "service", serv) klog.ErrorS(err, "Failed to sync endpoint for service", "serviceName", svcName, "virtualServer", serv)
} }
} else { } else {
klog.ErrorS(err, "Failed to sync service", "service", serv) klog.ErrorS(err, "Failed to sync service", "serviceName", svcName, "virtualServer", serv)
} }
} }
} }
@@ -1435,7 +1435,7 @@ func (proxier *Proxier) syncProxyRules() {
// For ports on node IPs, open the actual port and hold it. // For ports on node IPs, open the actual port and hold it.
for _, lp := range lps { for _, lp := range lps {
if proxier.portsMap[lp] != nil { if proxier.portsMap[lp] != nil {
klog.V(4).InfoS("Port was open before and is still needed", "port", lp.String()) klog.V(4).InfoS("Port was open before and is still needed", "port", lp)
replacementPortsMap[lp] = proxier.portsMap[lp] replacementPortsMap[lp] = proxier.portsMap[lp]
// We do not start listening on SCTP ports, according to our agreement in the // We do not start listening on SCTP ports, according to our agreement in the
// SCTP support KEP // SCTP support KEP
@@ -1451,10 +1451,10 @@ func (proxier *Proxier) syncProxyRules() {
UID: types.UID(proxier.hostname), UID: types.UID(proxier.hostname),
Namespace: "", Namespace: "",
}, nil, v1.EventTypeWarning, err.Error(), "SyncProxyRules", msg) }, nil, v1.EventTypeWarning, err.Error(), "SyncProxyRules", msg)
klog.ErrorS(err, "can't open port, skipping it", "port", lp.String()) klog.ErrorS(err, "Can't open port, skipping it", "port", lp)
continue continue
} }
klog.V(2).InfoS("Opened local port", "port", lp.String()) klog.V(2).InfoS("Opened local port", "port", lp)
if lp.Protocol == netutils.UDP { if lp.Protocol == netutils.UDP {
conntrack.ClearEntriesForPort(proxier.exec, lp.Port, isIPv6, v1.ProtocolUDP) conntrack.ClearEntriesForPort(proxier.exec, lp.Port, isIPv6, v1.ProtocolUDP)
@@ -1508,7 +1508,7 @@ func (proxier *Proxier) syncProxyRules() {
entryInvalidErr := false entryInvalidErr := false
for _, entry := range entries { for _, entry := range entries {
if valid := nodePortSet.validateEntry(entry); !valid { if valid := nodePortSet.validateEntry(entry); !valid {
klog.ErrorS(nil, "error adding entry to ipset", "entry", entry.String(), "ipset", nodePortSet.Name) klog.ErrorS(nil, "Error adding entry to ipset", "entry", entry, "ipset", nodePortSet.Name)
entryInvalidErr = true entryInvalidErr = true
break break
} }
@@ -1537,7 +1537,7 @@ func (proxier *Proxier) syncProxyRules() {
entryInvalidErr := false entryInvalidErr := false
for _, entry := range entries { for _, entry := range entries {
if valid := nodePortLocalSet.validateEntry(entry); !valid { if valid := nodePortLocalSet.validateEntry(entry); !valid {
klog.ErrorS(nil, "error adding entry to ipset", "entry", entry.String(), "ipset", nodePortLocalSet.Name) klog.ErrorS(nil, "Error adding entry to ipset", "entry", entry, "ipset", nodePortLocalSet.Name)
entryInvalidErr = true entryInvalidErr = true
break break
} }
@@ -1566,10 +1566,10 @@ func (proxier *Proxier) syncProxyRules() {
if err := proxier.syncService(svcNameString, serv, false, bindedAddresses); err == nil { if err := proxier.syncService(svcNameString, serv, false, bindedAddresses); err == nil {
activeIPVSServices[serv.String()] = true activeIPVSServices[serv.String()] = true
if err := proxier.syncEndpoint(svcName, svcInfo.NodeLocalExternal(), serv); err != nil { if err := proxier.syncEndpoint(svcName, svcInfo.NodeLocalExternal(), serv); err != nil {
klog.ErrorS(err, "Failed to sync endpoint for service", "service", serv) klog.ErrorS(err, "Failed to sync endpoint for service", "serviceName", svcName, "virtualServer", serv)
} }
} else { } else {
klog.ErrorS(err, "Failed to sync service", "service", serv) klog.ErrorS(err, "Failed to sync service", "serviceName", svcName, "virtualServer", serv)
} }
} }
} }
@@ -1584,7 +1584,7 @@ func (proxier *Proxier) syncProxyRules() {
} }
if valid := nodePortSet.validateEntry(entry); !valid { if valid := nodePortSet.validateEntry(entry); !valid {
klog.ErrorS(nil, "error adding entry to ipset", "entry", entry.String(), "ipset", nodePortSet.Name) klog.ErrorS(nil, "Error adding entry to ipset", "entry", entry, "ipset", nodePortSet.Name)
continue continue
} }
nodePortSet.activeEntries.Insert(entry.String()) nodePortSet.activeEntries.Insert(entry.String())
@@ -1671,7 +1671,7 @@ func (proxier *Proxier) syncProxyRules() {
// TODO: these could be made more consistent. // TODO: these could be made more consistent.
for _, svcIP := range staleServices.UnsortedList() { for _, svcIP := range staleServices.UnsortedList() {
if err := conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP); err != nil { if err := conntrack.ClearEntriesForIP(proxier.exec, svcIP, v1.ProtocolUDP); err != nil {
klog.ErrorS(err, "Failed to delete stale service IP connections", "ip", svcIP) klog.ErrorS(err, "Failed to delete stale service IP connections", "IP", svcIP)
} }
} }
proxier.deleteEndpointConnections(endpointUpdateResult.StaleEndpoints) proxier.deleteEndpointConnections(endpointUpdateResult.StaleEndpoints)
@@ -1958,18 +1958,18 @@ func (proxier *Proxier) deleteEndpointConnections(connectionMap []proxy.ServiceE
svcProto := svcInfo.Protocol() svcProto := svcInfo.Protocol()
err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIP().String(), endpointIP, svcProto) err := conntrack.ClearEntriesForNAT(proxier.exec, svcInfo.ClusterIP().String(), endpointIP, svcProto)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections", "servicePortName", epSvcPair.ServicePortName.String()) klog.ErrorS(err, "Failed to delete endpoint connections", "servicePortName", epSvcPair.ServicePortName)
} }
for _, extIP := range svcInfo.ExternalIPStrings() { for _, extIP := range svcInfo.ExternalIPStrings() {
err := conntrack.ClearEntriesForNAT(proxier.exec, extIP, endpointIP, svcProto) err := conntrack.ClearEntriesForNAT(proxier.exec, extIP, endpointIP, svcProto)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections for externalIP", "servicePortName", epSvcPair.ServicePortName.String(), "ip", extIP) klog.ErrorS(err, "Failed to delete endpoint connections for externalIP", "servicePortName", epSvcPair.ServicePortName, "IP", extIP)
} }
} }
for _, lbIP := range svcInfo.LoadBalancerIPStrings() { for _, lbIP := range svcInfo.LoadBalancerIPStrings() {
err := conntrack.ClearEntriesForNAT(proxier.exec, lbIP, endpointIP, svcProto) err := conntrack.ClearEntriesForNAT(proxier.exec, lbIP, endpointIP, svcProto)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to delete endpoint connections for LoadBalancerIP", "servicePortName", epSvcPair.ServicePortName.String(), "ip", lbIP) klog.ErrorS(err, "Failed to delete endpoint connections for LoadBalancerIP", "servicePortName", epSvcPair.ServicePortName, "IP", lbIP)
} }
} }
} }
@@ -1981,15 +1981,15 @@ func (proxier *Proxier) syncService(svcName string, vs *utilipvs.VirtualServer,
if appliedVirtualServer == nil || !appliedVirtualServer.Equal(vs) { if appliedVirtualServer == nil || !appliedVirtualServer.Equal(vs) {
if appliedVirtualServer == nil { if appliedVirtualServer == nil {
// IPVS service is not found, create a new service // IPVS service is not found, create a new service
klog.V(3).InfoS("Adding new service", "svcName", svcName, "address", fmt.Sprintf("%s:%d/%s", vs.Address, vs.Port, vs.Protocol)) klog.V(3).InfoS("Adding new service", "serviceName", svcName, "virtualServer", vs)
if err := proxier.ipvs.AddVirtualServer(vs); err != nil { if err := proxier.ipvs.AddVirtualServer(vs); err != nil {
klog.ErrorS(err, "Failed to add IPVS service", "svcName", svcName) klog.ErrorS(err, "Failed to add IPVS service", "serviceName", svcName)
return err return err
} }
} else { } else {
// IPVS service was changed, update the existing one // IPVS service was changed, update the existing one
// During updates, service VIP will not go down // During updates, service VIP will not go down
klog.V(3).InfoS("IPVS service was changed", "svcName", svcName) klog.V(3).InfoS("IPVS service was changed", "serviceName", svcName)
if err := proxier.ipvs.UpdateVirtualServer(vs); err != nil { if err := proxier.ipvs.UpdateVirtualServer(vs); err != nil {
klog.ErrorS(err, "Failed to update IPVS service") klog.ErrorS(err, "Failed to update IPVS service")
return err return err
@@ -2005,10 +2005,10 @@ func (proxier *Proxier) syncService(svcName string, vs *utilipvs.VirtualServer,
return nil return nil
} }
klog.V(4).InfoS("Bind addr", "address", vs.Address.String()) klog.V(4).InfoS("Bind address", "address", vs.Address)
_, err := proxier.netlinkHandle.EnsureAddressBind(vs.Address.String(), DefaultDummyDevice) _, err := proxier.netlinkHandle.EnsureAddressBind(vs.Address.String(), DefaultDummyDevice)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to bind service address to dummy device", "svcName", svcName) klog.ErrorS(err, "Failed to bind service address to dummy device", "serviceName", svcName)
return err return err
} }
} }
@@ -2019,7 +2019,7 @@ func (proxier *Proxier) syncService(svcName string, vs *utilipvs.VirtualServer,
func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNodeLocalEndpoints bool, vs *utilipvs.VirtualServer) error { func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNodeLocalEndpoints bool, vs *utilipvs.VirtualServer) error {
appliedVirtualServer, err := proxier.ipvs.GetVirtualServer(vs) appliedVirtualServer, err := proxier.ipvs.GetVirtualServer(vs)
if err != nil { if err != nil {
klog.Errorf("Failed to get IPVS service, error: %v", err) klog.ErrorS(err, "Failed to get IPVS service")
return err return err
} }
if appliedVirtualServer == nil { if appliedVirtualServer == nil {
@@ -2053,7 +2053,7 @@ func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNode
// externalTrafficPolicy=Local. // externalTrafficPolicy=Local.
svcInfo, ok := proxier.serviceMap[svcPortName] svcInfo, ok := proxier.serviceMap[svcPortName]
if !ok { if !ok {
klog.InfoS("Unable to filter endpoints due to missing Service info", "svcPortName", svcPortName) klog.InfoS("Unable to filter endpoints due to missing service info", "servicePortName", svcPortName)
} else { } else {
endpoints = proxy.FilterEndpoints(endpoints, svcInfo, proxier.nodeLabels) endpoints = proxy.FilterEndpoints(endpoints, svcInfo, proxier.nodeLabels)
} }
@@ -2108,7 +2108,7 @@ func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNode
if !proxier.gracefuldeleteManager.InTerminationList(uniqueRS) { if !proxier.gracefuldeleteManager.InTerminationList(uniqueRS) {
continue continue
} }
klog.V(5).InfoS("new ep is in graceful delete list", "uniqueRS", uniqueRS) klog.V(5).InfoS("new ep is in graceful delete list", "uniqueRealServer", uniqueRS)
err := proxier.gracefuldeleteManager.MoveRSOutofGracefulDeleteList(uniqueRS) err := proxier.gracefuldeleteManager.MoveRSOutofGracefulDeleteList(uniqueRS)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to delete endpoint in gracefulDeleteQueue", "endpoint", ep) klog.ErrorS(err, "Failed to delete endpoint in gracefulDeleteQueue", "endpoint", ep)
@@ -2144,10 +2144,10 @@ func (proxier *Proxier) syncEndpoint(svcPortName proxy.ServicePortName, onlyNode
Port: uint16(portNum), Port: uint16(portNum),
} }
klog.V(5).InfoS("Using graceful delete", "uniqueRS", uniqueRS) klog.V(5).InfoS("Using graceful delete", "uniqueRealServer", uniqueRS)
err = proxier.gracefuldeleteManager.GracefulDeleteRS(appliedVirtualServer, delDest) err = proxier.gracefuldeleteManager.GracefulDeleteRS(appliedVirtualServer, delDest)
if err != nil { if err != nil {
klog.ErrorS(err, "Failed to delete destination", "uniqueRS", uniqueRS) klog.ErrorS(err, "Failed to delete destination", "uniqueRealServer", uniqueRS)
continue continue
} }
} }
@@ -2166,9 +2166,9 @@ func (proxier *Proxier) cleanLegacyService(activeServices map[string]bool, curre
continue continue
} }
if _, ok := activeServices[cs]; !ok { if _, ok := activeServices[cs]; !ok {
klog.V(4).InfoS("Delete service", "service", svc.String()) klog.V(4).InfoS("Delete service", "virtualServer", svc)
if err := proxier.ipvs.DeleteVirtualServer(svc); err != nil { if err := proxier.ipvs.DeleteVirtualServer(svc); err != nil {
klog.ErrorS(err, "Failed to delete service", "service", svc.String()) klog.ErrorS(err, "Failed to delete service", "virtualServer", svc)
} }
addr := svc.Address.String() addr := svc.Address.String()
if _, ok := legacyBindAddrs[addr]; ok { if _, ok := legacyBindAddrs[addr]; ok {