istio

1
// Copyright Istio Authors
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14

15
// nolint: gocritic
16
package ambient
17

18
import (
19
	v1 "k8s.io/api/core/v1"
20

21
	networkingclient "istio.io/client-go/pkg/apis/networking/v1alpha3"
22
	"istio.io/istio/pilot/pkg/model"
23
	"istio.io/istio/pilot/pkg/serviceregistry/kube"
24
	"istio.io/istio/pkg/config"
25
	"istio.io/istio/pkg/config/schema/kind"
26
	"istio.io/istio/pkg/kube/krt"
27
	"istio.io/istio/pkg/log"
28
	"istio.io/istio/pkg/slices"
29
	"istio.io/istio/pkg/workloadapi"
30
)
31

32
func (a *index) ServicesCollection(
33
	Services krt.Collection[*v1.Service],
34
	ServiceEntries krt.Collection[*networkingclient.ServiceEntry],
35
	Waypoints krt.Collection[Waypoint],
36
	Namespaces krt.Collection[*v1.Namespace],
37
) krt.Collection[model.ServiceInfo] {
38
	ServicesInfo := krt.NewCollection(Services, func(ctx krt.HandlerContext, s *v1.Service) *model.ServiceInfo {
39
		portNames := map[int32]model.ServicePortName{}
40
		for _, p := range s.Spec.Ports {
41
			portNames[p.Port] = model.ServicePortName{
42
				PortName:       p.Name,
43
				TargetPortName: p.TargetPort.StrVal,
44
			}
45
		}
46
		waypoint := fetchWaypoint(ctx, Waypoints, Namespaces, s.ObjectMeta)
47
		a.networkUpdateTrigger.MarkDependant(ctx) // Mark we depend on out of band a.Network
48
		return &model.ServiceInfo{
49
			Service:       a.constructService(s, waypoint),
50
			PortNames:     portNames,
51
			LabelSelector: model.NewSelector(s.Spec.Selector),
52
			Source:        kind.Service,
53
		}
54
	}, krt.WithName("ServicesInfo"))
55
	ServiceEntriesInfo := krt.NewManyCollection(ServiceEntries, func(ctx krt.HandlerContext, s *networkingclient.ServiceEntry) []model.ServiceInfo {
56
		waypoint := fetchWaypoint(ctx, Waypoints, Namespaces, s.ObjectMeta)
57
		a.networkUpdateTrigger.MarkDependant(ctx) // Mark we depend on out of band a.Network
58
		return a.serviceEntriesInfo(s, waypoint)
59
	}, krt.WithName("ServiceEntriesInfo"))
60
	WorkloadServices := krt.JoinCollection([]krt.Collection[model.ServiceInfo]{ServicesInfo, ServiceEntriesInfo}, krt.WithName("WorkloadServices"))
61
	// workloadapi services NOT workloads x services somehow
62
	return WorkloadServices
63
}
64

65
func (a *index) serviceEntriesInfo(s *networkingclient.ServiceEntry, w *Waypoint) []model.ServiceInfo {
66
	sel := model.NewSelector(s.Spec.GetWorkloadSelector().GetLabels())
67
	portNames := map[int32]model.ServicePortName{}
68
	for _, p := range s.Spec.Ports {
69
		portNames[int32(p.Number)] = model.ServicePortName{
70
			PortName: p.Name,
71
		}
72
	}
73
	return slices.Map(a.constructServiceEntries(s, w), func(e *workloadapi.Service) model.ServiceInfo {
74
		return model.ServiceInfo{
75
			Service:       e,
76
			PortNames:     portNames,
77
			LabelSelector: sel,
78
			Source:        kind.ServiceEntry,
79
		}
80
	})
81
}
82

83
func (a *index) constructServiceEntries(svc *networkingclient.ServiceEntry, w *Waypoint) []*workloadapi.Service {
84
	addresses, err := slices.MapErr(svc.Spec.Addresses, a.toNetworkAddressFromCidr)
85
	if err != nil {
86
		// TODO: perhaps we should support CIDR in the future?
87
		return nil
88
	}
89
	ports := make([]*workloadapi.Port, 0, len(svc.Spec.Ports))
90
	for _, p := range svc.Spec.Ports {
91
		ports = append(ports, &workloadapi.Port{
92
			ServicePort: p.Number,
93
			TargetPort:  p.TargetPort,
94
		})
95
	}
96

97
	// handle svc waypoint scenario
98
	var waypointAddress *workloadapi.GatewayAddress
99
	if w != nil {
100
		waypointAddress = a.getWaypointAddress(w)
101
	}
102

103
	// TODO this is only checking one controller - we may be missing service vips for instances in another cluster
104
	res := make([]*workloadapi.Service, 0, len(svc.Spec.Hosts))
105
	for _, h := range svc.Spec.Hosts {
106
		res = append(res, &workloadapi.Service{
107
			Name:      svc.Name,
108
			Namespace: svc.Namespace,
109
			Hostname:  h,
110
			Addresses: addresses,
111
			Ports:     ports,
112
			Waypoint:  waypointAddress,
113
		})
114
	}
115
	return res
116
}
117

118
func (a *index) constructService(svc *v1.Service, w *Waypoint) *workloadapi.Service {
119
	ports := make([]*workloadapi.Port, 0, len(svc.Spec.Ports))
120
	for _, p := range svc.Spec.Ports {
121
		ports = append(ports, &workloadapi.Port{
122
			ServicePort: uint32(p.Port),
123
			TargetPort:  uint32(p.TargetPort.IntVal),
124
		})
125
	}
126

127
	addresses, err := slices.MapErr(getVIPs(svc), a.toNetworkAddress)
128
	if err != nil {
129
		log.Warnf("fail to parse service %v: %v", config.NamespacedName(svc), err)
130
		return nil
131
	}
132
	// handle svc waypoint scenario
133
	var waypointAddress *workloadapi.GatewayAddress
134
	if w != nil {
135
		waypointAddress = a.getWaypointAddress(w)
136
	}
137

138
	// TODO this is only checking one controller - we may be missing service vips for instances in another cluster
139
	return &workloadapi.Service{
140
		Name:      svc.Name,
141
		Namespace: svc.Namespace,
142
		Hostname:  string(kube.ServiceHostname(svc.Name, svc.Namespace, a.DomainSuffix)),
143
		Addresses: addresses,
144
		Ports:     ports,
145
		Waypoint:  waypointAddress,
146
	}
147
}
148

149
func getVIPs(svc *v1.Service) []string {
150
	res := []string{}
151
	if svc.Spec.ClusterIP != "" && svc.Spec.ClusterIP != v1.ClusterIPNone {
152
		res = append(res, svc.Spec.ClusterIP)
153
	}
154
	for _, ing := range svc.Status.LoadBalancer.Ingress {
155
		// IPs are strictly optional for loadbalancers - they may just have a hostname.
156
		if ing.IP != "" {
157
			res = append(res, ing.IP)
158
		}
159
	}
160
	return res
161
}
162