93860a5217
When switching from iptables or ipvs to nftables, clean up old iptables/ipvs rules. When switching the other way, clean up old nftables rules.
1358 lines
52 KiB
Go
1358 lines
52 KiB
Go
/*
|
|
Copyright 2015 The Kubernetes Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package nftables
|
|
|
|
//
|
|
// NOTE: this needs to be tested in e2e since it uses nftables for everything.
|
|
//
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base32"
|
|
"fmt"
|
|
"net"
|
|
"reflect"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/danwinship/knftables"
|
|
|
|
v1 "k8s.io/api/core/v1"
|
|
discovery "k8s.io/api/discovery/v1"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/sets"
|
|
"k8s.io/apimachinery/pkg/util/wait"
|
|
"k8s.io/client-go/tools/events"
|
|
utilsysctl "k8s.io/component-helpers/node/util/sysctl"
|
|
"k8s.io/klog/v2"
|
|
"k8s.io/kubernetes/pkg/proxy"
|
|
"k8s.io/kubernetes/pkg/proxy/conntrack"
|
|
"k8s.io/kubernetes/pkg/proxy/healthcheck"
|
|
"k8s.io/kubernetes/pkg/proxy/metaproxier"
|
|
"k8s.io/kubernetes/pkg/proxy/metrics"
|
|
proxyutil "k8s.io/kubernetes/pkg/proxy/util"
|
|
proxyutiliptables "k8s.io/kubernetes/pkg/proxy/util/iptables"
|
|
"k8s.io/kubernetes/pkg/util/async"
|
|
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
|
|
utiliptablestesting "k8s.io/kubernetes/pkg/util/iptables/testing"
|
|
utilexec "k8s.io/utils/exec"
|
|
netutils "k8s.io/utils/net"
|
|
)
|
|
|
|
const (
|
|
// the nftables table
|
|
kubeProxyTable = "kube-proxy"
|
|
|
|
// the services chain in the filter table
|
|
kubeServicesFilterChain = "KUBE-SERVICES"
|
|
|
|
// the services chain in the NAT table
|
|
kubeServicesChain = "KUBE-SERVICES"
|
|
|
|
// the external services chain
|
|
kubeExternalServicesChain = "KUBE-EXTERNAL-SERVICES"
|
|
|
|
// the nodeports chain
|
|
kubeNodePortsChain = "KUBE-NODEPORTS"
|
|
|
|
// the kubernetes postrouting chain
|
|
kubePostroutingChain = "KUBE-POSTROUTING"
|
|
|
|
// kubeMarkMasqChain is the mark-for-masquerade chain
|
|
kubeMarkMasqChain = "KUBE-MARK-MASQ"
|
|
|
|
// the kubernetes forward chain
|
|
kubeForwardChain = "KUBE-FORWARD"
|
|
|
|
// kubeProxyFirewallChain is the kube-proxy firewall chain
|
|
kubeProxyFirewallChain = "KUBE-PROXY-FIREWALL"
|
|
)
|
|
|
|
const sysctlNFConntrackTCPBeLiberal = "net/netfilter/nf_conntrack_tcp_be_liberal"
|
|
|
|
// internal struct for string service information
|
|
type servicePortInfo struct {
|
|
*proxy.BaseServicePortInfo
|
|
// The following fields are computed and stored for performance reasons.
|
|
nameString string
|
|
clusterPolicyChainName string
|
|
localPolicyChainName string
|
|
firewallChainName string
|
|
externalChainName string
|
|
}
|
|
|
|
// returns a new proxy.ServicePort which abstracts a serviceInfo
|
|
func newServiceInfo(port *v1.ServicePort, service *v1.Service, bsvcPortInfo *proxy.BaseServicePortInfo) proxy.ServicePort {
|
|
svcPort := &servicePortInfo{BaseServicePortInfo: bsvcPortInfo}
|
|
|
|
// Store the following for performance reasons.
|
|
svcName := types.NamespacedName{Namespace: service.Namespace, Name: service.Name}
|
|
svcPortName := proxy.ServicePortName{NamespacedName: svcName, Port: port.Name}
|
|
protocol := strings.ToLower(string(svcPort.Protocol()))
|
|
svcPort.nameString = svcPortName.String()
|
|
svcPort.clusterPolicyChainName = servicePortPolicyClusterChain(svcPort.nameString, protocol)
|
|
svcPort.localPolicyChainName = servicePortPolicyLocalChainName(svcPort.nameString, protocol)
|
|
svcPort.firewallChainName = serviceFirewallChainName(svcPort.nameString, protocol)
|
|
svcPort.externalChainName = serviceExternalChainName(svcPort.nameString, protocol)
|
|
|
|
return svcPort
|
|
}
|
|
|
|
// internal struct for endpoints information
|
|
type endpointInfo struct {
|
|
*proxy.BaseEndpointInfo
|
|
|
|
chainName string
|
|
}
|
|
|
|
// returns a new proxy.Endpoint which abstracts a endpointInfo
|
|
func newEndpointInfo(baseInfo *proxy.BaseEndpointInfo, svcPortName *proxy.ServicePortName) proxy.Endpoint {
|
|
return &endpointInfo{
|
|
BaseEndpointInfo: baseInfo,
|
|
chainName: servicePortEndpointChainName(svcPortName.String(), strings.ToLower(string(svcPortName.Protocol)), baseInfo.String()),
|
|
}
|
|
}
|
|
|
|
// Proxier is an nftables based proxy
|
|
type Proxier struct {
|
|
// ipFamily defines the IP family which this proxier is tracking.
|
|
ipFamily v1.IPFamily
|
|
|
|
// endpointsChanges and serviceChanges contains all changes to endpoints and
|
|
// services that happened since nftables was synced. For a single object,
|
|
// changes are accumulated, i.e. previous is state from before all of them,
|
|
// current is state after applying all of those.
|
|
endpointsChanges *proxy.EndpointsChangeTracker
|
|
serviceChanges *proxy.ServiceChangeTracker
|
|
|
|
mu sync.Mutex // protects the following fields
|
|
svcPortMap proxy.ServicePortMap
|
|
endpointsMap proxy.EndpointsMap
|
|
nodeLabels map[string]string
|
|
// endpointSlicesSynced, and servicesSynced are set to true
|
|
// when corresponding objects are synced after startup. This is used to avoid
|
|
// updating nftables with some partial data after kube-proxy restart.
|
|
endpointSlicesSynced bool
|
|
servicesSynced bool
|
|
initialized int32
|
|
syncRunner *async.BoundedFrequencyRunner // governs calls to syncProxyRules
|
|
syncPeriod time.Duration
|
|
|
|
// These are effectively const and do not need the mutex to be held.
|
|
iptables utiliptables.Interface
|
|
nftables knftables.Interface
|
|
masqueradeAll bool
|
|
masqueradeMark string
|
|
exec utilexec.Interface
|
|
localDetector proxyutiliptables.LocalTrafficDetector
|
|
hostname string
|
|
nodeIP net.IP
|
|
recorder events.EventRecorder
|
|
|
|
serviceHealthServer healthcheck.ServiceHealthServer
|
|
healthzServer *healthcheck.ProxierHealthServer
|
|
|
|
// Since converting probabilities (floats) to strings is expensive
|
|
// and we are using only probabilities in the format of 1/n, we are
|
|
// precomputing some number of those and cache for future reuse.
|
|
precomputedProbabilities []string
|
|
|
|
// The following buffers are used to reuse memory and avoid allocations
|
|
// that are significantly impacting performance.
|
|
iptablesData *bytes.Buffer
|
|
existingFilterChainsData *bytes.Buffer
|
|
filterChains proxyutil.LineBuffer
|
|
filterRules proxyutil.LineBuffer
|
|
natChains proxyutil.LineBuffer
|
|
natRules proxyutil.LineBuffer
|
|
|
|
// conntrackTCPLiberal indicates whether the system sets the kernel nf_conntrack_tcp_be_liberal
|
|
conntrackTCPLiberal bool
|
|
|
|
// nodePortAddresses selects the interfaces where nodePort works.
|
|
nodePortAddresses *proxyutil.NodePortAddresses
|
|
// networkInterfacer defines an interface for several net library functions.
|
|
// Inject for test purpose.
|
|
networkInterfacer proxyutil.NetworkInterfacer
|
|
}
|
|
|
|
// Proxier implements proxy.Provider
|
|
var _ proxy.Provider = &Proxier{}
|
|
|
|
// 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(ipFamily v1.IPFamily,
|
|
sysctl utilsysctl.Interface,
|
|
syncPeriod time.Duration,
|
|
minSyncPeriod time.Duration,
|
|
masqueradeAll bool,
|
|
masqueradeBit int,
|
|
localDetector proxyutiliptables.LocalTrafficDetector,
|
|
hostname string,
|
|
nodeIP net.IP,
|
|
recorder events.EventRecorder,
|
|
healthzServer *healthcheck.ProxierHealthServer,
|
|
nodePortAddressStrings []string,
|
|
initOnly bool,
|
|
) (*Proxier, error) {
|
|
nodePortAddresses := proxyutil.NewNodePortAddresses(ipFamily, nodePortAddressStrings)
|
|
|
|
// Be conservative in what you do, be liberal in what you accept from others.
|
|
// If it's non-zero, we mark only out of window RST segments as INVALID.
|
|
// Ref: https://docs.kernel.org/networking/nf_conntrack-sysctl.html
|
|
conntrackTCPLiberal := false
|
|
if val, err := sysctl.GetSysctl(sysctlNFConntrackTCPBeLiberal); err == nil && val != 0 {
|
|
conntrackTCPLiberal = true
|
|
klog.InfoS("nf_conntrack_tcp_be_liberal set, not installing DROP rules for INVALID packets")
|
|
}
|
|
|
|
if initOnly {
|
|
klog.InfoS("System initialized and --init-only specified")
|
|
return nil, nil
|
|
}
|
|
|
|
// Generate the masquerade mark to use for SNAT rules.
|
|
masqueradeValue := 1 << uint(masqueradeBit)
|
|
masqueradeMark := fmt.Sprintf("%#08x", masqueradeValue)
|
|
klog.V(2).InfoS("Using nftables mark for masquerade", "ipFamily", ipFamily, "mark", masqueradeMark)
|
|
|
|
serviceHealthServer := healthcheck.NewServiceHealthServer(hostname, recorder, nodePortAddresses, healthzServer)
|
|
|
|
var nftablesFamily knftables.Family
|
|
if ipFamily == v1.IPv4Protocol {
|
|
nftablesFamily = knftables.IPv4Family
|
|
} else {
|
|
nftablesFamily = knftables.IPv6Family
|
|
}
|
|
nft, err := knftables.New(nftablesFamily, kubeProxyTable)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
proxier := &Proxier{
|
|
ipFamily: ipFamily,
|
|
svcPortMap: make(proxy.ServicePortMap),
|
|
serviceChanges: proxy.NewServiceChangeTracker(newServiceInfo, ipFamily, recorder, nil),
|
|
endpointsMap: make(proxy.EndpointsMap),
|
|
endpointsChanges: proxy.NewEndpointsChangeTracker(hostname, newEndpointInfo, ipFamily, recorder, nil),
|
|
syncPeriod: syncPeriod,
|
|
iptables: utiliptablestesting.NewFake(),
|
|
nftables: nft,
|
|
masqueradeAll: masqueradeAll,
|
|
masqueradeMark: masqueradeMark,
|
|
exec: utilexec.New(),
|
|
localDetector: localDetector,
|
|
hostname: hostname,
|
|
nodeIP: nodeIP,
|
|
recorder: recorder,
|
|
serviceHealthServer: serviceHealthServer,
|
|
healthzServer: healthzServer,
|
|
precomputedProbabilities: make([]string, 0, 1001),
|
|
iptablesData: bytes.NewBuffer(nil),
|
|
existingFilterChainsData: bytes.NewBuffer(nil),
|
|
filterChains: proxyutil.NewLineBuffer(),
|
|
filterRules: proxyutil.NewLineBuffer(),
|
|
natChains: proxyutil.NewLineBuffer(),
|
|
natRules: proxyutil.NewLineBuffer(),
|
|
nodePortAddresses: nodePortAddresses,
|
|
networkInterfacer: proxyutil.RealNetwork{},
|
|
conntrackTCPLiberal: conntrackTCPLiberal,
|
|
}
|
|
|
|
burstSyncs := 2
|
|
klog.V(2).InfoS("NFTables sync params", "ipFamily", ipFamily, "minSyncPeriod", minSyncPeriod, "syncPeriod", syncPeriod, "burstSyncs", burstSyncs)
|
|
proxier.syncRunner = async.NewBoundedFrequencyRunner("sync-runner", proxier.syncProxyRules, minSyncPeriod, syncPeriod, burstSyncs)
|
|
|
|
return proxier, nil
|
|
}
|
|
|
|
// NewDualStackProxier creates a MetaProxier instance, with IPv4 and IPv6 proxies.
|
|
func NewDualStackProxier(
|
|
sysctl utilsysctl.Interface,
|
|
syncPeriod time.Duration,
|
|
minSyncPeriod time.Duration,
|
|
masqueradeAll bool,
|
|
masqueradeBit int,
|
|
localDetectors [2]proxyutiliptables.LocalTrafficDetector,
|
|
hostname string,
|
|
nodeIPs map[v1.IPFamily]net.IP,
|
|
recorder events.EventRecorder,
|
|
healthzServer *healthcheck.ProxierHealthServer,
|
|
nodePortAddresses []string,
|
|
initOnly bool,
|
|
) (proxy.Provider, error) {
|
|
// Create an ipv4 instance of the single-stack proxier
|
|
ipv4Proxier, err := NewProxier(v1.IPv4Protocol, sysctl,
|
|
syncPeriod, minSyncPeriod, masqueradeAll, masqueradeBit, localDetectors[0], hostname,
|
|
nodeIPs[v1.IPv4Protocol], recorder, healthzServer, nodePortAddresses, initOnly)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to create ipv4 proxier: %v", err)
|
|
}
|
|
|
|
ipv6Proxier, err := NewProxier(v1.IPv6Protocol, sysctl,
|
|
syncPeriod, minSyncPeriod, masqueradeAll, masqueradeBit, localDetectors[1], hostname,
|
|
nodeIPs[v1.IPv6Protocol], recorder, healthzServer, nodePortAddresses, initOnly)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("unable to create ipv6 proxier: %v", err)
|
|
}
|
|
if initOnly {
|
|
return nil, nil
|
|
}
|
|
return metaproxier.NewMetaProxier(ipv4Proxier, ipv6Proxier), nil
|
|
}
|
|
|
|
type iptablesJumpChain struct {
|
|
table utiliptables.Table
|
|
dstChain utiliptables.Chain
|
|
srcChain utiliptables.Chain
|
|
comment string
|
|
extraArgs []string
|
|
}
|
|
|
|
var iptablesJumpChains = []iptablesJumpChain{
|
|
{utiliptables.TableFilter, kubeExternalServicesChain, utiliptables.ChainInput, "kubernetes externally-visible service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeExternalServicesChain, utiliptables.ChainForward, "kubernetes externally-visible service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeServicesFilterChain, utiliptables.ChainForward, "kubernetes service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeServicesFilterChain, utiliptables.ChainOutput, "kubernetes service portals", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeForwardChain, utiliptables.ChainForward, "kubernetes forwarding rules", nil},
|
|
{utiliptables.TableFilter, kubeProxyFirewallChain, utiliptables.ChainInput, "kubernetes load balancer firewall", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeProxyFirewallChain, utiliptables.ChainOutput, "kubernetes load balancer firewall", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableFilter, kubeProxyFirewallChain, utiliptables.ChainForward, "kubernetes load balancer firewall", []string{"-m", "conntrack", "--ctstate", "NEW"}},
|
|
{utiliptables.TableNAT, kubeServicesChain, utiliptables.ChainOutput, "kubernetes service portals", nil},
|
|
{utiliptables.TableNAT, kubeServicesChain, utiliptables.ChainPrerouting, "kubernetes service portals", nil},
|
|
{utiliptables.TableNAT, kubePostroutingChain, utiliptables.ChainPostrouting, "kubernetes postrouting rules", nil},
|
|
}
|
|
|
|
// CleanupLeftovers removes all nftables rules and chains created by the Proxier
|
|
// It returns true if an error was encountered. Errors are logged.
|
|
func CleanupLeftovers() bool {
|
|
var encounteredError bool
|
|
|
|
for _, family := range []knftables.Family{knftables.IPv4Family, knftables.IPv6Family} {
|
|
nft, err := knftables.New(family, kubeProxyTable)
|
|
if err == nil {
|
|
tx := nft.NewTransaction()
|
|
tx.Delete(&knftables.Table{})
|
|
err = nft.Run(context.TODO(), tx)
|
|
}
|
|
if err != nil && !knftables.IsNotFound(err) {
|
|
klog.ErrorS(err, "Error cleaning up nftables rules")
|
|
encounteredError = true
|
|
}
|
|
}
|
|
|
|
return encounteredError
|
|
}
|
|
|
|
func computeProbability(n int) string {
|
|
return fmt.Sprintf("%0.10f", 1.0/float64(n))
|
|
}
|
|
|
|
// This assumes proxier.mu is held
|
|
func (proxier *Proxier) precomputeProbabilities(numberOfPrecomputed int) {
|
|
if len(proxier.precomputedProbabilities) == 0 {
|
|
proxier.precomputedProbabilities = append(proxier.precomputedProbabilities, "<bad value>")
|
|
}
|
|
for i := len(proxier.precomputedProbabilities); i <= numberOfPrecomputed; i++ {
|
|
proxier.precomputedProbabilities = append(proxier.precomputedProbabilities, computeProbability(i))
|
|
}
|
|
}
|
|
|
|
// This assumes proxier.mu is held
|
|
func (proxier *Proxier) probability(n int) string {
|
|
if n >= len(proxier.precomputedProbabilities) {
|
|
proxier.precomputeProbabilities(n)
|
|
}
|
|
return proxier.precomputedProbabilities[n]
|
|
}
|
|
|
|
// Sync is called to synchronize the proxier state to nftables as soon as possible.
|
|
func (proxier *Proxier) Sync() {
|
|
if proxier.healthzServer != nil {
|
|
proxier.healthzServer.QueuedUpdate(proxier.ipFamily)
|
|
}
|
|
metrics.SyncProxyRulesLastQueuedTimestamp.SetToCurrentTime()
|
|
proxier.syncRunner.Run()
|
|
}
|
|
|
|
// 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() {
|
|
// Update healthz timestamp at beginning in case Sync() never succeeds.
|
|
if proxier.healthzServer != nil {
|
|
proxier.healthzServer.Updated(proxier.ipFamily)
|
|
}
|
|
|
|
// synthesize "last change queued" time as the informers are syncing.
|
|
metrics.SyncProxyRulesLastQueuedTimestamp.SetToCurrentTime()
|
|
proxier.syncRunner.Loop(wait.NeverStop)
|
|
}
|
|
|
|
func (proxier *Proxier) setInitialized(value bool) {
|
|
var initialized int32
|
|
if value {
|
|
initialized = 1
|
|
}
|
|
atomic.StoreInt32(&proxier.initialized, initialized)
|
|
}
|
|
|
|
func (proxier *Proxier) isInitialized() bool {
|
|
return atomic.LoadInt32(&proxier.initialized) > 0
|
|
}
|
|
|
|
// OnServiceAdd is called whenever creation of new service object
|
|
// is observed.
|
|
func (proxier *Proxier) OnServiceAdd(service *v1.Service) {
|
|
proxier.OnServiceUpdate(nil, service)
|
|
}
|
|
|
|
// OnServiceUpdate is called whenever modification of an existing
|
|
// service object is observed.
|
|
func (proxier *Proxier) OnServiceUpdate(oldService, service *v1.Service) {
|
|
if proxier.serviceChanges.Update(oldService, service) && proxier.isInitialized() {
|
|
proxier.Sync()
|
|
}
|
|
}
|
|
|
|
// OnServiceDelete is called whenever deletion of an existing service
|
|
// object is observed.
|
|
func (proxier *Proxier) OnServiceDelete(service *v1.Service) {
|
|
proxier.OnServiceUpdate(service, nil)
|
|
|
|
}
|
|
|
|
// OnServiceSynced is called once all the initial event handlers were
|
|
// called and the state is fully propagated to local cache.
|
|
func (proxier *Proxier) OnServiceSynced() {
|
|
proxier.mu.Lock()
|
|
proxier.servicesSynced = true
|
|
proxier.setInitialized(proxier.endpointSlicesSynced)
|
|
proxier.mu.Unlock()
|
|
|
|
// Sync unconditionally - this is called once per lifetime.
|
|
proxier.syncProxyRules()
|
|
}
|
|
|
|
// OnEndpointSliceAdd is called whenever creation of a new endpoint slice object
|
|
// is observed.
|
|
func (proxier *Proxier) OnEndpointSliceAdd(endpointSlice *discovery.EndpointSlice) {
|
|
if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() {
|
|
proxier.Sync()
|
|
}
|
|
}
|
|
|
|
// OnEndpointSliceUpdate is called whenever modification of an existing endpoint
|
|
// slice object is observed.
|
|
func (proxier *Proxier) OnEndpointSliceUpdate(_, endpointSlice *discovery.EndpointSlice) {
|
|
if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, false) && proxier.isInitialized() {
|
|
proxier.Sync()
|
|
}
|
|
}
|
|
|
|
// OnEndpointSliceDelete is called whenever deletion of an existing endpoint slice
|
|
// object is observed.
|
|
func (proxier *Proxier) OnEndpointSliceDelete(endpointSlice *discovery.EndpointSlice) {
|
|
if proxier.endpointsChanges.EndpointSliceUpdate(endpointSlice, true) && proxier.isInitialized() {
|
|
proxier.Sync()
|
|
}
|
|
}
|
|
|
|
// OnEndpointSlicesSynced is called once all the initial event handlers were
|
|
// called and the state is fully propagated to local cache.
|
|
func (proxier *Proxier) OnEndpointSlicesSynced() {
|
|
proxier.mu.Lock()
|
|
proxier.endpointSlicesSynced = true
|
|
proxier.setInitialized(proxier.servicesSynced)
|
|
proxier.mu.Unlock()
|
|
|
|
// Sync unconditionally - this is called once per lifetime.
|
|
proxier.syncProxyRules()
|
|
}
|
|
|
|
// OnNodeAdd is called whenever creation of new node object
|
|
// is observed.
|
|
func (proxier *Proxier) OnNodeAdd(node *v1.Node) {
|
|
if node.Name != proxier.hostname {
|
|
klog.ErrorS(nil, "Received a watch event for a node that doesn't match the current node",
|
|
"eventNode", node.Name, "currentNode", proxier.hostname)
|
|
return
|
|
}
|
|
|
|
if reflect.DeepEqual(proxier.nodeLabels, node.Labels) {
|
|
return
|
|
}
|
|
|
|
proxier.mu.Lock()
|
|
proxier.nodeLabels = map[string]string{}
|
|
for k, v := range node.Labels {
|
|
proxier.nodeLabels[k] = v
|
|
}
|
|
proxier.mu.Unlock()
|
|
klog.V(4).InfoS("Updated proxier node labels", "labels", node.Labels)
|
|
|
|
proxier.Sync()
|
|
}
|
|
|
|
// OnNodeUpdate is called whenever modification of an existing
|
|
// node object is observed.
|
|
func (proxier *Proxier) OnNodeUpdate(oldNode, node *v1.Node) {
|
|
if node.Name != proxier.hostname {
|
|
klog.ErrorS(nil, "Received a watch event for a node that doesn't match the current node",
|
|
"eventNode", node.Name, "currentNode", proxier.hostname)
|
|
return
|
|
}
|
|
|
|
if reflect.DeepEqual(proxier.nodeLabels, node.Labels) {
|
|
return
|
|
}
|
|
|
|
proxier.mu.Lock()
|
|
proxier.nodeLabels = map[string]string{}
|
|
for k, v := range node.Labels {
|
|
proxier.nodeLabels[k] = v
|
|
}
|
|
proxier.mu.Unlock()
|
|
klog.V(4).InfoS("Updated proxier node labels", "labels", node.Labels)
|
|
|
|
proxier.Sync()
|
|
}
|
|
|
|
// OnNodeDelete is called whenever deletion of an existing node
|
|
// object is observed.
|
|
func (proxier *Proxier) OnNodeDelete(node *v1.Node) {
|
|
if node.Name != proxier.hostname {
|
|
klog.ErrorS(nil, "Received a watch event for a node that doesn't match the current node",
|
|
"eventNode", node.Name, "currentNode", proxier.hostname)
|
|
return
|
|
}
|
|
|
|
proxier.mu.Lock()
|
|
proxier.nodeLabels = nil
|
|
proxier.mu.Unlock()
|
|
|
|
proxier.Sync()
|
|
}
|
|
|
|
// OnNodeSynced is called once all the initial event handlers were
|
|
// called and the state is fully propagated to local cache.
|
|
func (proxier *Proxier) OnNodeSynced() {
|
|
}
|
|
|
|
// portProtoHash takes the ServicePortName and protocol for a service
|
|
// returns the associated 16 character hash. This is computed by hashing (sha256)
|
|
// then encoding to base32 and truncating to 16 chars. We do this because IPTables
|
|
// Chain Names must be <= 28 chars long, and the longer they are the harder they are to read.
|
|
func portProtoHash(servicePortName string, protocol string) string {
|
|
hash := sha256.Sum256([]byte(servicePortName + protocol))
|
|
encoded := base32.StdEncoding.EncodeToString(hash[:])
|
|
return encoded[:16]
|
|
}
|
|
|
|
const (
|
|
servicePortPolicyClusterChainNamePrefix = "KUBE-SVC-"
|
|
servicePortPolicyLocalChainNamePrefix = "KUBE-SVL-"
|
|
serviceFirewallChainNamePrefix = "KUBE-FW-"
|
|
serviceExternalChainNamePrefix = "KUBE-EXT-"
|
|
servicePortEndpointChainNamePrefix = "KUBE-SEP-"
|
|
)
|
|
|
|
// servicePortPolicyClusterChain returns the name of the KUBE-SVC-XXXX chain for a service, which is the
|
|
// main iptables chain for that service, used for dispatching to endpoints when using `Cluster`
|
|
// traffic policy.
|
|
func servicePortPolicyClusterChain(servicePortName string, protocol string) string {
|
|
return servicePortPolicyClusterChainNamePrefix + portProtoHash(servicePortName, protocol)
|
|
}
|
|
|
|
// servicePortPolicyLocalChainName returns the name of the KUBE-SVL-XXXX chain for a service, which
|
|
// handles dispatching to local endpoints when using `Local` traffic policy. This chain only
|
|
// exists if the service has `Local` internal or external traffic policy.
|
|
func servicePortPolicyLocalChainName(servicePortName string, protocol string) string {
|
|
return servicePortPolicyLocalChainNamePrefix + portProtoHash(servicePortName, protocol)
|
|
}
|
|
|
|
// serviceFirewallChainName returns the name of the KUBE-FW-XXXX chain for a service, which
|
|
// is used to implement the filtering for the LoadBalancerSourceRanges feature.
|
|
func serviceFirewallChainName(servicePortName string, protocol string) string {
|
|
return serviceFirewallChainNamePrefix + portProtoHash(servicePortName, protocol)
|
|
}
|
|
|
|
// serviceExternalChainName returns the name of the KUBE-EXT-XXXX chain for a service, which
|
|
// implements "short-circuiting" for internally-originated external-destination traffic when using
|
|
// `Local` external traffic policy. It forwards traffic from local sources to the KUBE-SVC-XXXX
|
|
// chain and traffic from external sources to the KUBE-SVL-XXXX chain.
|
|
func serviceExternalChainName(servicePortName string, protocol string) string {
|
|
return serviceExternalChainNamePrefix + portProtoHash(servicePortName, protocol)
|
|
}
|
|
|
|
// servicePortEndpointChainName returns the name of the KUBE-SEP-XXXX chain for a particular
|
|
// service endpoint.
|
|
func servicePortEndpointChainName(servicePortName string, protocol string, endpoint string) string {
|
|
hash := sha256.Sum256([]byte(servicePortName + protocol + endpoint))
|
|
encoded := base32.StdEncoding.EncodeToString(hash[:])
|
|
return servicePortEndpointChainNamePrefix + encoded[:16]
|
|
}
|
|
|
|
func isServiceChainName(chainString string) bool {
|
|
prefixes := []string{
|
|
servicePortPolicyClusterChainNamePrefix,
|
|
servicePortPolicyLocalChainNamePrefix,
|
|
servicePortEndpointChainNamePrefix,
|
|
serviceFirewallChainNamePrefix,
|
|
serviceExternalChainNamePrefix,
|
|
}
|
|
|
|
for _, p := range prefixes {
|
|
if strings.HasPrefix(chainString, p) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// This is where all of the nftables calls happen.
|
|
// This assumes proxier.mu is NOT held
|
|
func (proxier *Proxier) syncProxyRules() {
|
|
proxier.mu.Lock()
|
|
defer proxier.mu.Unlock()
|
|
|
|
// don't sync rules till we've received services and endpoints
|
|
if !proxier.isInitialized() {
|
|
klog.V(2).InfoS("Not syncing nftables until Services and Endpoints have been received from master")
|
|
return
|
|
}
|
|
|
|
// Keep track of how long syncs take.
|
|
start := time.Now()
|
|
defer func() {
|
|
metrics.SyncProxyRulesLatency.Observe(metrics.SinceInSeconds(start))
|
|
klog.V(2).InfoS("SyncProxyRules complete", "elapsed", time.Since(start))
|
|
}()
|
|
|
|
serviceUpdateResult := proxier.svcPortMap.Update(proxier.serviceChanges)
|
|
endpointUpdateResult := proxier.endpointsMap.Update(proxier.endpointsChanges)
|
|
|
|
klog.V(2).InfoS("Syncing nftables rules")
|
|
|
|
success := false
|
|
defer func() {
|
|
if !success {
|
|
klog.InfoS("Sync failed", "retryingTime", proxier.syncPeriod)
|
|
proxier.syncRunner.RetryAfter(proxier.syncPeriod)
|
|
}
|
|
}()
|
|
|
|
// Ensure that our jump rules (eg from PREROUTING to KUBE-SERVICES) exist.
|
|
// We can't do this as part of the iptables-restore because we don't want
|
|
// to specify/replace *all* of the rules in PREROUTING, etc.
|
|
for _, jump := range iptablesJumpChains {
|
|
if _, err := proxier.iptables.EnsureChain(jump.table, jump.dstChain); err != nil {
|
|
klog.ErrorS(err, "Failed to ensure chain exists", "table", jump.table, "chain", jump.dstChain)
|
|
return
|
|
}
|
|
args := jump.extraArgs
|
|
if jump.comment != "" {
|
|
args = append(args, "-m", "comment", "--comment", jump.comment)
|
|
}
|
|
args = append(args, "-j", string(jump.dstChain))
|
|
if _, err := proxier.iptables.EnsureRule(utiliptables.Prepend, jump.table, jump.srcChain, args...); err != nil {
|
|
klog.ErrorS(err, "Failed to ensure chain jumps", "table", jump.table, "srcChain", jump.srcChain, "dstChain", jump.dstChain)
|
|
return
|
|
}
|
|
}
|
|
|
|
//
|
|
// Below this point we will not return until we try to write the nftables rules.
|
|
//
|
|
|
|
// Reset all buffers used later.
|
|
// This is to avoid memory reallocations and thus improve performance.
|
|
proxier.filterChains.Reset()
|
|
proxier.filterRules.Reset()
|
|
proxier.natChains.Reset()
|
|
proxier.natRules.Reset()
|
|
|
|
// Write chain lines for all the "top-level" chains we'll be filling in
|
|
for _, chainName := range []utiliptables.Chain{kubeServicesFilterChain, kubeExternalServicesChain, kubeForwardChain, kubeProxyFirewallChain} {
|
|
proxier.filterChains.Write(utiliptables.MakeChainLine(chainName))
|
|
}
|
|
for _, chainName := range []utiliptables.Chain{kubeServicesChain, kubeNodePortsChain, kubePostroutingChain, kubeMarkMasqChain} {
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(chainName))
|
|
}
|
|
|
|
// Install the kubernetes-specific postrouting rules. We use a whole chain for
|
|
// this so that it is easier to flush and change, for example if the mark
|
|
// value should ever change.
|
|
|
|
proxier.natRules.Write(
|
|
"-A", string(kubePostroutingChain),
|
|
"-m", "mark", "!", "--mark", fmt.Sprintf("%s/%s", proxier.masqueradeMark, proxier.masqueradeMark),
|
|
"-j", "RETURN",
|
|
)
|
|
// Clear the mark to avoid re-masquerading if the packet re-traverses the network stack.
|
|
proxier.natRules.Write(
|
|
"-A", string(kubePostroutingChain),
|
|
"-j", "MARK", "--xor-mark", proxier.masqueradeMark,
|
|
)
|
|
proxier.natRules.Write(
|
|
"-A", string(kubePostroutingChain),
|
|
"-m", "comment", "--comment", `"kubernetes service traffic requiring SNAT"`,
|
|
"-j", "MASQUERADE", "--random-fully",
|
|
)
|
|
|
|
// Install the kubernetes-specific masquerade mark rule. We use a whole chain for
|
|
// this so that it is easier to flush and change, for example if the mark
|
|
// value should ever change.
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeMarkMasqChain),
|
|
"-j", "MARK", "--or-mark", proxier.masqueradeMark,
|
|
)
|
|
|
|
// Accumulate NAT chains to keep.
|
|
activeNATChains := sets.New[string]()
|
|
|
|
// Compute total number of endpoint chains across all services
|
|
// to get a sense of how big the cluster is.
|
|
totalEndpoints := 0
|
|
for svcName := range proxier.svcPortMap {
|
|
totalEndpoints += len(proxier.endpointsMap[svcName])
|
|
}
|
|
|
|
// These two variables are used to publish the sync_proxy_rules_no_endpoints_total
|
|
// metric.
|
|
serviceNoLocalEndpointsTotalInternal := 0
|
|
serviceNoLocalEndpointsTotalExternal := 0
|
|
|
|
// Build rules for each service-port.
|
|
for svcName, svc := range proxier.svcPortMap {
|
|
svcInfo, ok := svc.(*servicePortInfo)
|
|
if !ok {
|
|
klog.ErrorS(nil, "Failed to cast serviceInfo", "serviceName", svcName)
|
|
continue
|
|
}
|
|
protocol := strings.ToLower(string(svcInfo.Protocol()))
|
|
svcPortNameString := svcInfo.nameString
|
|
|
|
// Figure out the endpoints for Cluster and Local traffic policy.
|
|
// allLocallyReachableEndpoints is the set of all endpoints that can be routed to
|
|
// from this node, given the service's traffic policies. hasEndpoints is true
|
|
// if the service has any usable endpoints on any node, not just this one.
|
|
allEndpoints := proxier.endpointsMap[svcName]
|
|
clusterEndpoints, localEndpoints, allLocallyReachableEndpoints, hasEndpoints := proxy.CategorizeEndpoints(allEndpoints, svcInfo, proxier.nodeLabels)
|
|
|
|
// Note the endpoint chains that will be used
|
|
for _, ep := range allLocallyReachableEndpoints {
|
|
if epInfo, ok := ep.(*endpointInfo); ok {
|
|
activeNATChains.Insert(epInfo.chainName)
|
|
}
|
|
}
|
|
|
|
// clusterPolicyChain contains the endpoints used with "Cluster" traffic policy
|
|
clusterPolicyChain := svcInfo.clusterPolicyChainName
|
|
usesClusterPolicyChain := len(clusterEndpoints) > 0 && svcInfo.UsesClusterEndpoints()
|
|
if usesClusterPolicyChain {
|
|
activeNATChains.Insert(clusterPolicyChain)
|
|
}
|
|
|
|
// localPolicyChain contains the endpoints used with "Local" traffic policy
|
|
localPolicyChain := svcInfo.localPolicyChainName
|
|
usesLocalPolicyChain := len(localEndpoints) > 0 && svcInfo.UsesLocalEndpoints()
|
|
if usesLocalPolicyChain {
|
|
activeNATChains.Insert(localPolicyChain)
|
|
}
|
|
|
|
// internalPolicyChain is the chain containing the endpoints for
|
|
// "internal" (ClusterIP) traffic. internalTrafficChain is the chain that
|
|
// internal traffic is routed to (which is always the same as
|
|
// internalPolicyChain). hasInternalEndpoints is true if we should
|
|
// generate rules pointing to internalTrafficChain, or false if there are
|
|
// no available internal endpoints.
|
|
internalPolicyChain := clusterPolicyChain
|
|
hasInternalEndpoints := hasEndpoints
|
|
if svcInfo.InternalPolicyLocal() {
|
|
internalPolicyChain = localPolicyChain
|
|
if len(localEndpoints) == 0 {
|
|
hasInternalEndpoints = false
|
|
}
|
|
}
|
|
internalTrafficChain := internalPolicyChain
|
|
|
|
// Similarly, externalPolicyChain is the chain containing the endpoints
|
|
// for "external" (NodePort, LoadBalancer, and ExternalIP) traffic.
|
|
// externalTrafficChain is the chain that external traffic is routed to
|
|
// (which is always the service's "EXT" chain). hasExternalEndpoints is
|
|
// true if there are endpoints that will be reached by external traffic.
|
|
// (But we may still have to generate externalTrafficChain even if there
|
|
// are no external endpoints, to ensure that the short-circuit rules for
|
|
// local traffic are set up.)
|
|
externalPolicyChain := clusterPolicyChain
|
|
hasExternalEndpoints := hasEndpoints
|
|
if svcInfo.ExternalPolicyLocal() {
|
|
externalPolicyChain = localPolicyChain
|
|
if len(localEndpoints) == 0 {
|
|
hasExternalEndpoints = false
|
|
}
|
|
}
|
|
externalTrafficChain := svcInfo.externalChainName // eventually jumps to externalPolicyChain
|
|
|
|
// usesExternalTrafficChain is based on hasEndpoints, not hasExternalEndpoints,
|
|
// because we need the local-traffic-short-circuiting rules even when there
|
|
// are no externally-usable endpoints.
|
|
usesExternalTrafficChain := hasEndpoints && svcInfo.ExternallyAccessible()
|
|
if usesExternalTrafficChain {
|
|
activeNATChains.Insert(externalTrafficChain)
|
|
}
|
|
|
|
// Traffic to LoadBalancer IPs can go directly to externalTrafficChain
|
|
// unless LoadBalancerSourceRanges is in use in which case we will
|
|
// create a firewall chain.
|
|
loadBalancerTrafficChain := externalTrafficChain
|
|
fwChain := svcInfo.firewallChainName
|
|
usesFWChain := hasEndpoints && len(svcInfo.LoadBalancerVIPStrings()) > 0 && len(svcInfo.LoadBalancerSourceRanges()) > 0
|
|
if usesFWChain {
|
|
activeNATChains.Insert(fwChain)
|
|
loadBalancerTrafficChain = fwChain
|
|
}
|
|
|
|
var internalTrafficFilterTarget, internalTrafficFilterComment string
|
|
var externalTrafficFilterTarget, externalTrafficFilterComment string
|
|
if !hasEndpoints {
|
|
// The service has no endpoints at all; hasInternalEndpoints and
|
|
// hasExternalEndpoints will also be false, and we will not
|
|
// generate any chains in the "nat" table for the service; only
|
|
// rules in the "filter" table rejecting incoming packets for
|
|
// the service's IPs.
|
|
internalTrafficFilterTarget = "REJECT"
|
|
internalTrafficFilterComment = fmt.Sprintf(`"%s has no endpoints"`, svcPortNameString)
|
|
externalTrafficFilterTarget = "REJECT"
|
|
externalTrafficFilterComment = internalTrafficFilterComment
|
|
} else {
|
|
if !hasInternalEndpoints {
|
|
// The internalTrafficPolicy is "Local" but there are no local
|
|
// endpoints. Traffic to the clusterIP will be dropped, but
|
|
// external traffic may still be accepted.
|
|
internalTrafficFilterTarget = "DROP"
|
|
internalTrafficFilterComment = fmt.Sprintf(`"%s has no local endpoints"`, svcPortNameString)
|
|
serviceNoLocalEndpointsTotalInternal++
|
|
}
|
|
if !hasExternalEndpoints {
|
|
// The externalTrafficPolicy is "Local" but there are no
|
|
// local endpoints. Traffic to "external" IPs from outside
|
|
// the cluster will be dropped, but traffic from inside
|
|
// the cluster may still be accepted.
|
|
externalTrafficFilterTarget = "DROP"
|
|
externalTrafficFilterComment = fmt.Sprintf(`"%s has no local endpoints"`, svcPortNameString)
|
|
serviceNoLocalEndpointsTotalExternal++
|
|
}
|
|
}
|
|
|
|
// Capture the clusterIP.
|
|
if hasInternalEndpoints {
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeServicesChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcPortNameString),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", svcInfo.ClusterIP().String(),
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", string(internalTrafficChain))
|
|
} else {
|
|
// No endpoints.
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeServicesFilterChain),
|
|
"-m", "comment", "--comment", internalTrafficFilterComment,
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", svcInfo.ClusterIP().String(),
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", internalTrafficFilterTarget,
|
|
)
|
|
}
|
|
|
|
// Capture externalIPs.
|
|
for _, externalIP := range svcInfo.ExternalIPStrings() {
|
|
if hasEndpoints {
|
|
// Send traffic bound for external IPs to the "external
|
|
// destinations" chain.
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeServicesChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s external IP"`, svcPortNameString),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", externalIP,
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", string(externalTrafficChain))
|
|
}
|
|
if !hasExternalEndpoints {
|
|
// Either no endpoints at all (REJECT) or no endpoints for
|
|
// external traffic (DROP anything that didn't get
|
|
// short-circuited by the EXT chain.)
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeExternalServicesChain),
|
|
"-m", "comment", "--comment", externalTrafficFilterComment,
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", externalIP,
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", externalTrafficFilterTarget,
|
|
)
|
|
}
|
|
}
|
|
|
|
// Capture load-balancer ingress.
|
|
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
|
|
if hasEndpoints {
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeServicesChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcPortNameString),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", lbip,
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", string(loadBalancerTrafficChain))
|
|
|
|
}
|
|
if usesFWChain {
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeProxyFirewallChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s traffic not accepted by %s"`, svcPortNameString, svcInfo.firewallChainName),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", lbip,
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", "DROP")
|
|
}
|
|
}
|
|
if !hasExternalEndpoints {
|
|
// Either no endpoints at all (REJECT) or no endpoints for
|
|
// external traffic (DROP anything that didn't get short-circuited
|
|
// by the EXT chain.)
|
|
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeExternalServicesChain),
|
|
"-m", "comment", "--comment", externalTrafficFilterComment,
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", lbip,
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", externalTrafficFilterTarget,
|
|
)
|
|
}
|
|
}
|
|
|
|
// Capture nodeports.
|
|
if svcInfo.NodePort() != 0 {
|
|
if hasEndpoints {
|
|
// Jump to the external destination chain. For better or for
|
|
// worse, nodeports are not subect to loadBalancerSourceRanges,
|
|
// and we can't change that.
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeNodePortsChain),
|
|
"-m", "comment", "--comment", svcPortNameString,
|
|
"-m", protocol, "-p", protocol,
|
|
"--dport", strconv.Itoa(svcInfo.NodePort()),
|
|
"-j", string(externalTrafficChain))
|
|
}
|
|
if !hasExternalEndpoints {
|
|
// Either no endpoints at all (REJECT) or no endpoints for
|
|
// external traffic (DROP anything that didn't get
|
|
// short-circuited by the EXT chain.)
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeExternalServicesChain),
|
|
"-m", "comment", "--comment", externalTrafficFilterComment,
|
|
"-m", "addrtype", "--dst-type", "LOCAL",
|
|
"-m", protocol, "-p", protocol,
|
|
"--dport", strconv.Itoa(svcInfo.NodePort()),
|
|
"-j", externalTrafficFilterTarget,
|
|
)
|
|
}
|
|
}
|
|
|
|
// Set up internal traffic handling.
|
|
if hasInternalEndpoints {
|
|
if proxier.masqueradeAll {
|
|
proxier.natRules.Write(
|
|
"-A", string(internalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcPortNameString),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", svcInfo.ClusterIP().String(),
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
"-j", string(kubeMarkMasqChain))
|
|
} else if proxier.localDetector.IsImplemented() {
|
|
// This masquerades off-cluster traffic to a service VIP. The
|
|
// idea is that you can establish a static route for your
|
|
// Service range, routing to any node, and that node will
|
|
// bridge into the Service for you. Since that might bounce
|
|
// off-node, we masquerade here.
|
|
proxier.natRules.Write(
|
|
"-A", string(internalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s cluster IP"`, svcPortNameString),
|
|
"-m", protocol, "-p", protocol,
|
|
"-d", svcInfo.ClusterIP().String(),
|
|
"--dport", strconv.Itoa(svcInfo.Port()),
|
|
proxier.localDetector.IfNotLocal(),
|
|
"-j", string(kubeMarkMasqChain))
|
|
}
|
|
}
|
|
|
|
// Set up external traffic handling (if any "external" destinations are
|
|
// enabled). All captured traffic for all external destinations should
|
|
// jump to externalTrafficChain, which will handle some special cases and
|
|
// then jump to externalPolicyChain.
|
|
if usesExternalTrafficChain {
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(utiliptables.Chain(externalTrafficChain)))
|
|
|
|
if !svcInfo.ExternalPolicyLocal() {
|
|
// If we are using non-local endpoints we need to masquerade,
|
|
// in case we cross nodes.
|
|
proxier.natRules.Write(
|
|
"-A", string(externalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"masquerade traffic for %s external destinations"`, svcPortNameString),
|
|
"-j", string(kubeMarkMasqChain))
|
|
} else {
|
|
// If we are only using same-node endpoints, we can retain the
|
|
// source IP in most cases.
|
|
|
|
if proxier.localDetector.IsImplemented() {
|
|
// Treat all locally-originated pod -> external destination
|
|
// traffic as a special-case. It is subject to neither
|
|
// form of traffic policy, which simulates going up-and-out
|
|
// to an external load-balancer and coming back in.
|
|
proxier.natRules.Write(
|
|
"-A", string(externalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"pod traffic for %s external destinations"`, svcPortNameString),
|
|
proxier.localDetector.IfLocal(),
|
|
"-j", string(clusterPolicyChain))
|
|
}
|
|
|
|
// Locally originated traffic (not a pod, but the host node)
|
|
// still needs masquerade because the LBIP itself is a local
|
|
// address, so that will be the chosen source IP.
|
|
proxier.natRules.Write(
|
|
"-A", string(externalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"masquerade LOCAL traffic for %s external destinations"`, svcPortNameString),
|
|
"-m", "addrtype", "--src-type", "LOCAL",
|
|
"-j", string(kubeMarkMasqChain))
|
|
|
|
// Redirect all src-type=LOCAL -> external destination to the
|
|
// policy=cluster chain. This allows traffic originating
|
|
// from the host to be redirected to the service correctly.
|
|
proxier.natRules.Write(
|
|
"-A", string(externalTrafficChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"route LOCAL traffic for %s external destinations"`, svcPortNameString),
|
|
"-m", "addrtype", "--src-type", "LOCAL",
|
|
"-j", string(clusterPolicyChain))
|
|
}
|
|
|
|
// Anything else falls thru to the appropriate policy chain.
|
|
if hasExternalEndpoints {
|
|
proxier.natRules.Write(
|
|
"-A", string(externalTrafficChain),
|
|
"-j", string(externalPolicyChain))
|
|
}
|
|
}
|
|
|
|
// Set up firewall chain, if needed
|
|
if usesFWChain {
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(utiliptables.Chain(fwChain)))
|
|
|
|
// The service firewall rules are created based on the
|
|
// loadBalancerSourceRanges field. This only works for VIP-like
|
|
// loadbalancers that preserve source IPs. For loadbalancers which
|
|
// direct traffic to service NodePort, the firewall rules will not
|
|
// apply.
|
|
|
|
// firewall filter based on each source range
|
|
allowFromNode := false
|
|
for _, src := range svcInfo.LoadBalancerSourceRanges() {
|
|
proxier.natRules.Write(
|
|
"-A", string(fwChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcPortNameString),
|
|
"-s", src,
|
|
"-j", string(externalTrafficChain),
|
|
)
|
|
_, cidr, err := netutils.ParseCIDRSloppy(src)
|
|
if err != nil {
|
|
klog.ErrorS(err, "Error parsing CIDR in LoadBalancerSourceRanges, dropping it", "cidr", cidr)
|
|
} else if cidr.Contains(proxier.nodeIP) {
|
|
allowFromNode = true
|
|
}
|
|
}
|
|
// For VIP-like LBs, the VIP is often added as a local
|
|
// address (via an IP route rule). In that case, a request
|
|
// from a node to the VIP will not hit the loadbalancer but
|
|
// will loop back with the source IP set to the VIP. We
|
|
// need the following rules to allow requests from this node.
|
|
if allowFromNode {
|
|
for _, lbip := range svcInfo.LoadBalancerVIPStrings() {
|
|
proxier.natRules.Write(
|
|
"-A", string(fwChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"%s loadbalancer IP"`, svcPortNameString),
|
|
"-s", lbip,
|
|
"-j", string(externalTrafficChain),
|
|
)
|
|
}
|
|
}
|
|
// If the packet was able to reach the end of firewall chain,
|
|
// then it did not get DNATed, so it will match the
|
|
// corresponding KUBE-PROXY-FIREWALL rule.
|
|
proxier.natRules.Write(
|
|
"-A", string(fwChain),
|
|
"-m", "comment", "--comment", fmt.Sprintf(`"other traffic to %s will be dropped by KUBE-PROXY-FIREWALL"`, svcPortNameString),
|
|
)
|
|
}
|
|
|
|
// If Cluster policy is in use, create the chain and create rules jumping
|
|
// from clusterPolicyChain to the clusterEndpoints
|
|
if usesClusterPolicyChain {
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(utiliptables.Chain(clusterPolicyChain)))
|
|
proxier.writeServiceToEndpointRules(svcPortNameString, svcInfo, clusterPolicyChain, clusterEndpoints)
|
|
}
|
|
|
|
// If Local policy is in use, create the chain and create rules jumping
|
|
// from localPolicyChain to the localEndpoints
|
|
if usesLocalPolicyChain {
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(utiliptables.Chain(localPolicyChain)))
|
|
proxier.writeServiceToEndpointRules(svcPortNameString, svcInfo, localPolicyChain, localEndpoints)
|
|
}
|
|
|
|
// Generate the per-endpoint chains.
|
|
for _, ep := range allLocallyReachableEndpoints {
|
|
epInfo, ok := ep.(*endpointInfo)
|
|
if !ok {
|
|
klog.ErrorS(nil, "Failed to cast endpointInfo", "endpointInfo", ep)
|
|
continue
|
|
}
|
|
|
|
endpointChain := epInfo.chainName
|
|
|
|
// Create the endpoint chain
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(utiliptables.Chain(endpointChain)))
|
|
activeNATChains.Insert(endpointChain)
|
|
|
|
// Handle traffic that loops back to the originator with SNAT.
|
|
proxier.natRules.Write(
|
|
"-A", string(endpointChain),
|
|
"-m", "comment", "--comment", svcPortNameString,
|
|
"-s", epInfo.IP(),
|
|
"-j", string(kubeMarkMasqChain),
|
|
)
|
|
commentAndAffinityArgs := []string{"-m", "comment", "--comment", svcPortNameString}
|
|
if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
|
|
commentAndAffinityArgs = append(commentAndAffinityArgs, "-m", "recent", "--name", string(endpointChain), "--set")
|
|
}
|
|
// DNAT to final destination.
|
|
proxier.natRules.Write(
|
|
"-A", string(endpointChain),
|
|
commentAndAffinityArgs,
|
|
"-m", protocol, "-p", protocol,
|
|
"-j", "DNAT", "--to-destination", epInfo.String(),
|
|
)
|
|
}
|
|
}
|
|
|
|
// Delete chains no longer in use.
|
|
deletedChains := 0
|
|
var existingNATChains map[utiliptables.Chain]struct{}
|
|
|
|
proxier.iptablesData.Reset()
|
|
if err := proxier.iptables.SaveInto(utiliptables.TableNAT, proxier.iptablesData); err == nil {
|
|
existingNATChains = utiliptables.GetChainsFromTable(proxier.iptablesData.Bytes())
|
|
|
|
for chain := range existingNATChains {
|
|
if !activeNATChains.Has(string(chain)) {
|
|
chainString := string(chain)
|
|
if !isServiceChainName(chainString) {
|
|
// Ignore chains that aren't ours.
|
|
continue
|
|
}
|
|
// We must (as per iptables) write a chain-line
|
|
// for it, which has the nice effect of flushing
|
|
// the chain. Then we can remove the chain.
|
|
proxier.natChains.Write(utiliptables.MakeChainLine(chain))
|
|
proxier.natRules.Write("-X", chainString)
|
|
deletedChains++
|
|
}
|
|
}
|
|
} else {
|
|
klog.ErrorS(err, "Failed to execute iptables-save: stale chains will not be deleted")
|
|
}
|
|
|
|
// Finally, tail-call to the nodePorts chain. This needs to be after all
|
|
// other service portal rules.
|
|
if proxier.nodePortAddresses.MatchAll() {
|
|
isIPv6 := proxier.ipFamily == v1.IPv6Protocol
|
|
|
|
destinations := []string{"-m", "addrtype", "--dst-type", "LOCAL"}
|
|
// Block localhost nodePorts
|
|
if isIPv6 {
|
|
destinations = append(destinations, "!", "-d", "::1/128")
|
|
} else {
|
|
destinations = append(destinations, "!", "-d", "127.0.0.0/8")
|
|
}
|
|
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeServicesChain),
|
|
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
|
|
destinations,
|
|
"-j", string(kubeNodePortsChain))
|
|
} else {
|
|
nodeIPs, err := proxier.nodePortAddresses.GetNodeIPs(proxier.networkInterfacer)
|
|
if err != nil {
|
|
klog.ErrorS(err, "Failed to get node ip address matching nodeport cidrs, services with nodeport may not work as intended", "CIDRs", proxier.nodePortAddresses)
|
|
}
|
|
for _, ip := range nodeIPs {
|
|
if ip.IsLoopback() {
|
|
klog.ErrorS(nil, "--nodeport-addresses includes localhost but localhost NodePorts are not supported", "address", ip.String())
|
|
continue
|
|
}
|
|
|
|
// create nodeport rules for each IP one by one
|
|
proxier.natRules.Write(
|
|
"-A", string(kubeServicesChain),
|
|
"-m", "comment", "--comment", `"kubernetes service nodeports; NOTE: this must be the last rule in this chain"`,
|
|
"-d", ip.String(),
|
|
"-j", string(kubeNodePortsChain))
|
|
}
|
|
}
|
|
|
|
// Drop the packets in INVALID state, which would potentially cause
|
|
// unexpected connection reset if nf_conntrack_tcp_be_liberal is not set.
|
|
// Ref: https://github.com/kubernetes/kubernetes/issues/74839
|
|
// Ref: https://github.com/kubernetes/kubernetes/issues/117924
|
|
if !proxier.conntrackTCPLiberal {
|
|
proxier.filterRules.Write(
|
|
"-A", string(kubeForwardChain),
|
|
"-m", "conntrack",
|
|
"--ctstate", "INVALID",
|
|
"-j", "DROP",
|
|
)
|
|
}
|
|
|
|
metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableFilter)).Set(float64(proxier.filterRules.Lines()))
|
|
metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableNAT)).Set(float64(proxier.natRules.Lines() - deletedChains))
|
|
|
|
// Sync rules.
|
|
proxier.iptablesData.Reset()
|
|
proxier.iptablesData.WriteString("*filter\n")
|
|
proxier.iptablesData.Write(proxier.filterChains.Bytes())
|
|
proxier.iptablesData.Write(proxier.filterRules.Bytes())
|
|
proxier.iptablesData.WriteString("COMMIT\n")
|
|
proxier.iptablesData.WriteString("*nat\n")
|
|
proxier.iptablesData.Write(proxier.natChains.Bytes())
|
|
proxier.iptablesData.Write(proxier.natRules.Bytes())
|
|
proxier.iptablesData.WriteString("COMMIT\n")
|
|
|
|
klog.V(2).InfoS("Reloading service nftables data",
|
|
"numServices", len(proxier.svcPortMap),
|
|
"numEndpoints", totalEndpoints,
|
|
"numFilterChains", proxier.filterChains.Lines(),
|
|
"numFilterRules", proxier.filterRules.Lines(),
|
|
"numNATChains", proxier.natChains.Lines(),
|
|
"numNATRules", proxier.natRules.Lines(),
|
|
)
|
|
klog.V(9).InfoS("Restoring iptables", "rules", proxier.iptablesData.Bytes())
|
|
|
|
// NOTE: NoFlushTables is used so we don't flush non-kubernetes chains in the table
|
|
err := proxier.iptables.RestoreAll(proxier.iptablesData.Bytes(), utiliptables.NoFlushTables, utiliptables.RestoreCounters)
|
|
if err != nil {
|
|
if pErr, ok := err.(utiliptables.ParseError); ok {
|
|
lines := utiliptables.ExtractLines(proxier.iptablesData.Bytes(), pErr.Line(), 3)
|
|
klog.ErrorS(pErr, "Failed to execute iptables-restore", "rules", lines)
|
|
} else {
|
|
klog.ErrorS(err, "Failed to execute iptables-restore")
|
|
}
|
|
metrics.IptablesRestoreFailuresTotal.Inc()
|
|
return
|
|
}
|
|
success = true
|
|
|
|
for name, lastChangeTriggerTimes := range endpointUpdateResult.LastChangeTriggerTimes {
|
|
for _, lastChangeTriggerTime := range lastChangeTriggerTimes {
|
|
latency := metrics.SinceInSeconds(lastChangeTriggerTime)
|
|
metrics.NetworkProgrammingLatency.Observe(latency)
|
|
klog.V(4).InfoS("Network programming", "endpoint", klog.KRef(name.Namespace, name.Name), "elapsed", latency)
|
|
}
|
|
}
|
|
|
|
metrics.SyncProxyRulesNoLocalEndpointsTotal.WithLabelValues("internal").Set(float64(serviceNoLocalEndpointsTotalInternal))
|
|
metrics.SyncProxyRulesNoLocalEndpointsTotal.WithLabelValues("external").Set(float64(serviceNoLocalEndpointsTotalExternal))
|
|
if proxier.healthzServer != nil {
|
|
proxier.healthzServer.Updated(proxier.ipFamily)
|
|
}
|
|
metrics.SyncProxyRulesLastTimestamp.SetToCurrentTime()
|
|
|
|
// Update service healthchecks. The endpoints list might include services that are
|
|
// not "OnlyLocal", but the services list will not, and the serviceHealthServer
|
|
// will just drop those endpoints.
|
|
if err := proxier.serviceHealthServer.SyncServices(proxier.svcPortMap.HealthCheckNodePorts()); err != nil {
|
|
klog.ErrorS(err, "Error syncing healthcheck services")
|
|
}
|
|
if err := proxier.serviceHealthServer.SyncEndpoints(proxier.endpointsMap.LocalReadyEndpoints()); err != nil {
|
|
klog.ErrorS(err, "Error syncing healthcheck endpoints")
|
|
}
|
|
|
|
// Finish housekeeping, clear stale conntrack entries for UDP Services
|
|
conntrack.CleanStaleEntries(proxier.ipFamily == v1.IPv6Protocol, proxier.exec, proxier.svcPortMap, serviceUpdateResult, endpointUpdateResult)
|
|
}
|
|
|
|
func (proxier *Proxier) writeServiceToEndpointRules(svcPortNameString string, svcInfo proxy.ServicePort, svcChain string, endpoints []proxy.Endpoint) {
|
|
// First write session affinity rules, if applicable.
|
|
if svcInfo.SessionAffinityType() == v1.ServiceAffinityClientIP {
|
|
for _, ep := range endpoints {
|
|
epInfo, ok := ep.(*endpointInfo)
|
|
if !ok {
|
|
continue
|
|
}
|
|
comment := fmt.Sprintf(`"%s -> %s"`, svcPortNameString, epInfo.String())
|
|
proxier.natRules.Write(
|
|
"-A", string(svcChain),
|
|
"-m", "comment", "--comment", comment,
|
|
"-m", "recent", "--name", string(epInfo.chainName),
|
|
"--rcheck", "--seconds", strconv.Itoa(svcInfo.StickyMaxAgeSeconds()), "--reap",
|
|
"-j", string(epInfo.chainName),
|
|
)
|
|
}
|
|
}
|
|
|
|
// Now write loadbalancing rules.
|
|
numEndpoints := len(endpoints)
|
|
for i, ep := range endpoints {
|
|
epInfo, ok := ep.(*endpointInfo)
|
|
if !ok {
|
|
continue
|
|
}
|
|
comment := fmt.Sprintf(`"%s -> %s"`, svcPortNameString, epInfo.String())
|
|
|
|
if i < (numEndpoints - 1) {
|
|
// Each rule is a probabilistic match.
|
|
proxier.natRules.Write(
|
|
"-A", string(svcChain),
|
|
"-m", "comment", "--comment", comment,
|
|
"-m", "statistic",
|
|
"--mode", "random",
|
|
"--probability", proxier.probability(numEndpoints-i),
|
|
"-j", string(epInfo.chainName),
|
|
)
|
|
} else {
|
|
// The final (or only if n == 1) rule is a guaranteed match.
|
|
proxier.natRules.Write(
|
|
"-A", string(svcChain),
|
|
"-m", "comment", "--comment", comment,
|
|
"-j", string(epInfo.chainName),
|
|
)
|
|
}
|
|
}
|
|
}
|