const kubeProxyTable … const filterPreroutingChain … const filterInputChain … const filterForwardChain … const filterOutputChain … const filterOutputPostDNATChain … const natPreroutingChain … const natOutputChain … const natPostroutingChain … const servicesChain … const serviceIPsMap … const serviceNodePortsMap … const nodePortIPsSet … const clusterIPsSet … const serviceEndpointsCheckChain … const nodePortEndpointsCheckChain … const noEndpointServicesMap … const noEndpointNodePortsMap … const rejectChain … const clusterIPsCheckChain … const firewallIPsMap … const firewallCheckChain … const markMasqChain … const masqueradingChain … // NewDualStackProxier creates a MetaProxier instance, with IPv4 and IPv6 proxies. func NewDualStackProxier( ctx context.Context, syncPeriod time.Duration, minSyncPeriod time.Duration, masqueradeAll bool, masqueradeBit int, localDetectors map[v1.IPFamily]proxyutil.LocalTrafficDetector, hostname string, nodeIPs map[v1.IPFamily]net.IP, recorder events.EventRecorder, healthzServer *healthcheck.ProxierHealthServer, nodePortAddresses []string, initOnly bool, ) (proxy.Provider, error) { … } type Proxier … var _ … // NewProxier returns a new nftables Proxier. Once a proxier is created, it will keep // nftables up to date in the background and will not terminate if a particular nftables // call fails. func NewProxier(ctx context.Context, ipFamily v1.IPFamily, syncPeriod time.Duration, minSyncPeriod time.Duration, masqueradeAll bool, masqueradeBit int, localDetector proxyutil.LocalTrafficDetector, hostname string, nodeIP net.IP, recorder events.EventRecorder, healthzServer *healthcheck.ProxierHealthServer, nodePortAddressStrings []string, initOnly bool, ) (*Proxier, error) { … } // Create a knftables.Interface and check if we can use the nftables proxy mode on this host. func getNFTablesInterface(ipFamily v1.IPFamily) (knftables.Interface, error) { … } type servicePortInfo … // returns a new proxy.ServicePort which abstracts a serviceInfo func newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort { … } type endpointInfo … // returns a new proxy.Endpoint which abstracts a endpointInfo func newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, svcPortName *proxy.ServicePortName) proxy.Endpoint { … } type nftablesBaseChain … var nftablesBaseChains … type nftablesJumpChain … var nftablesJumpChains … // ensureChain adds commands to tx to ensure that chain exists and doesn't contain // anything from before this transaction (using createdChains to ensure that we don't // Flush a chain more than once and lose *new* rules as well.) // If skipCreation is true, chain will not be added to the transaction, but will be added to the createdChains // for proper cleanup in the end of the sync iteration. func ensureChain(chain string, tx *knftables.Transaction, createdChains sets.Set[string], skipCreation bool) { … } func (proxier *Proxier) setupNFTables(tx *knftables.Transaction) { … } // CleanupLeftovers removes all nftables rules and chains created by the Proxier // It returns true if an error was encountered. Errors are logged. func CleanupLeftovers(ctx context.Context) bool { … } // Sync is called to synchronize the proxier state to nftables as soon as possible. func (proxier *Proxier) Sync() { … } // SyncLoop runs periodic work. This is expected to run as a goroutine or as the main loop of the app. It does not return. func (proxier *Proxier) SyncLoop() { … } func (proxier *Proxier) setInitialized(value bool) { … } func (proxier *Proxier) isInitialized() bool { … } // OnServiceAdd is called whenever creation of new service object // is observed. func (proxier *Proxier) OnServiceAdd(service *v1.Service) { … } // OnServiceUpdate is called whenever modification of an existing // service object is observed. func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) { … } // OnServiceDelete is called whenever deletion of an existing service // object is observed. func (proxier *Proxier) OnServiceDelete(service *v1.Service) { … } // OnServiceSynced is called once all the initial event handlers were // called and the state is fully propagated to local cache. func (proxier *Proxier) OnServiceSynced() { … } // OnEndpointSliceAdd is called whenever creation of a new endpoint slice object // is observed. func (proxier *Proxier) OnEndpointSliceAdd(endpointSlice *discovery.EndpointSlice) { … } // OnEndpointSliceUpdate is called whenever modification of an existing endpoint // slice object is observed. func (proxier *Proxier) OnEndpointSliceUpdate(_, endpointSlice *discovery.EndpointSlice) { … } // OnEndpointSliceDelete is called whenever deletion of an existing endpoint slice // object is observed. func (proxier *Proxier) OnEndpointSliceDelete(endpointSlice *discovery.EndpointSlice) { … } // OnEndpointSlicesSynced is called once all the initial event handlers were // called and the state is fully propagated to local cache. func (proxier *Proxier) OnEndpointSlicesSynced() { … } // OnNodeAdd is called whenever creation of new node object // is observed. func (proxier *Proxier) OnNodeAdd(node *v1.Node) { … } // OnNodeUpdate is called whenever modification of an existing // node object is observed. func (proxier *Proxier) OnNodeUpdate(oldNode, node *v1.Node) { … } // OnNodeDelete is called whenever deletion of an existing node // object is observed. func (proxier *Proxier) OnNodeDelete(node *v1.Node) { … } // OnNodeSynced is called once all the initial event handlers were // called and the state is fully propagated to local cache. func (proxier *Proxier) OnNodeSynced() { … } // OnServiceCIDRsChanged is called whenever a change is observed // in any of the ServiceCIDRs, and provides complete list of service cidrs. func (proxier *Proxier) OnServiceCIDRsChanged(cidrs []string) { … } const chainNamePrefixLengthMax … const chainNameBaseLengthMax … const servicePortPolicyClusterChainNamePrefix … const servicePortPolicyLocalChainNamePrefix … const serviceExternalChainNamePrefix … const servicePortEndpointChainNamePrefix … const servicePortEndpointAffinityNamePrefix … const servicePortFirewallChainNamePrefix … // hashAndTruncate prefixes name with a hash of itself and then truncates to // chainNameBaseLengthMax. The hash ensures that (a) the name is still unique if we have // to truncate the end, and (b) it's visually distinguishable from other chains that would // otherwise have nearly identical names (e.g., different endpoint chains for a given // service that differ in only a single digit). func hashAndTruncate(name string) string { … } // servicePortChainNameBase returns the base name for a chain for the given ServicePort. // This is something like "HASH-namespace/serviceName/protocol/portName", e.g, // "ULMVA6XW-ns1/svc1/tcp/p80". func servicePortChainNameBase(servicePortName *proxy.ServicePortName, protocol string) string { … } // servicePortEndpointChainNameBase returns the suffix for chain names for the given // endpoint. This is something like // "HASH-namespace/serviceName/protocol/portName__endpointIP/endpointport", e.g., // "5OJB2KTY-ns1/svc1/tcp/p80__10.180.0.1/80". func servicePortEndpointChainNameBase(servicePortName *proxy.ServicePortName, protocol, endpoint string) string { … } func isServiceChainName(chainString string) bool { … } func isAffinitySetName(set string) bool { … } type nftElementStorage … // joinNFTSlice converts nft element key or value (type []string) to string to store in the nftElementStorage. // The separator is the same as the one used by nft commands, so we know that the parsing is going to be unambiguous. func joinNFTSlice(k []string) string { … } // splitNFTSlice converts nftElementStorage key or value string representation back to slice. func splitNFTSlice(k string) []string { … } // newNFTElementStorage creates an empty nftElementStorage. // nftElementStorage.reset() must be called before the first usage. func newNFTElementStorage(containerType, containerName string) *nftElementStorage { … } // reset clears the internal state and flushes the nftables map/set. func (s *nftElementStorage) reset(tx *knftables.Transaction) { … } // resetLeftoverKeys is only called internally by nftElementStorage methods. func (s *nftElementStorage) resetLeftoverKeys() { … } // ensureElem adds elem to the transaction if elem is not present in the container, and updates internal // leftoverKeys set to track unused elements. func (s *nftElementStorage) ensureElem(tx *knftables.Transaction, elem *knftables.Element) { … } func (s *nftElementStorage) cleanupLeftoverKeys(tx *knftables.Transaction) { … } // This is where all of the nftables calls happen. // This assumes proxier.mu is NOT held func (proxier *Proxier) syncProxyRules() { … } func (proxier *Proxier) writeServiceToEndpointRules(tx *knftables.Transaction, svcInfo *servicePortInfo, svcChain string, endpoints []proxy.Endpoint) { … }
Codebase Browser
kubernetes