Class: Async::GRPC::XDS::ControlPlane

Inherits:
Object
  • Object
show all
Defined in:
lib/async/grpc/xds/control_plane.rb

Overview

Maintains xDS resource snapshots and notifies ADS streams when resources change.

Constant Summary collapse

CLUSTER_TYPE = 
ResourceBuilder::CLUSTER_TYPE
ENDPOINT_TYPE = 
ResourceBuilder::ENDPOINT_TYPE

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(identifier: "async-grpc-xds") ⇒ ControlPlane

Initialize an empty control plane.



26
27
28
29
30
31
32
 
# File 'lib/async/grpc/xds/control_plane.rb', line 26

def initialize(identifier: "async-grpc-xds")
	@identifier = identifier
	@resources = Hash.new{|hash, type_url| hash[type_url] = {}}
	@versions = Hash.new(0)
	@streams = Set.new.compare_by_identity
	@mutex = Mutex.new
end

Instance Attribute Details

#identifier ⇒ Object (readonly)

Returns the value of attribute identifier.



34
35
36
 
# File 'lib/async/grpc/xds/control_plane.rb', line 34

def identifier
  @identifier
end

Instance Method Details

#register_stream(stream) ⇒ Object

Register a stream to receive resource-change notifications.



153
154
155
156
157
 
# File 'lib/async/grpc/xds/control_plane.rb', line 153

def register_stream(stream)
	@mutex.synchronize do
		@streams.add(stream)
	end
end

#remove_cluster(name) ⇒ Object

Remove a cluster resource.



58
59
60
 
# File 'lib/async/grpc/xds/control_plane.rb', line 58

def remove_cluster(name)
	remove_resource(CLUSTER_TYPE, name.to_s)
end

#remove_endpoints(cluster_name) ⇒ Object

Remove the endpoint assignment for a cluster.



64
65
66
 
# File 'lib/async/grpc/xds/control_plane.rb', line 64

def remove_endpoints(cluster_name)
	remove_resource(ENDPOINT_TYPE, cluster_name.to_s)
end

#remove_resource(type_url, name) ⇒ Object

Remove an xDS resource and notify subscribed streams.



87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/async/grpc/xds/control_plane.rb', line 87

def remove_resource(type_url, name)
	notify = false
	
	@mutex.synchronize do
		if @resources[type_url].delete(name)
			@versions[type_url] += 1
			notify = true
		end
	end
	
	notify_streams(type_url) if notify
end

#remove_stream(stream) ⇒ Object

Remove a registered stream.



161
162
163
164
165
 
# File 'lib/async/grpc/xds/control_plane.rb', line 161

def remove_stream(stream)
	@mutex.synchronize do
		@streams.delete(stream)
	end
end

#resource_names(type_url) ⇒ Object

Get the available resource names for a type.



103
104
105
106
107
 
# File 'lib/async/grpc/xds/control_plane.rb', line 103

def resource_names(type_url)
	@mutex.synchronize do
		@resources[type_url].keys
	end
end

#resources(type_url, names = nil) ⇒ Object

Get resources of a given type.



113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/async/grpc/xds/control_plane.rb', line 113

def resources(type_url, names = nil)
	@mutex.synchronize do
		resources = @resources[type_url]
		
		if names && names.any?
			names.filter_map{|name| resources[name]}
		else
			resources.values
		end
	end
end

#response(type_url, names = nil) ⇒ Object

Build a discovery response for a resource type.



138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/async/grpc/xds/control_plane.rb', line 138

def response(type_url, names = nil)
	resources = self.resources(type_url, names)
	version = self.version(type_url)
	
	Envoy::Service::Discovery::V3::DiscoveryResponse.new(
		version_info: version,
		resources: resources.map{|resource| ResourceBuilder.pack(resource)},
		type_url: type_url,
		nonce: "#{type_url}:#{version}:#{SecureRandom.hex(8)}",
		control_plane: Envoy::Config::Core::V3::ControlPlane.new(identifier: @identifier)
	)
end

#update_cluster(name, resource = nil, **options) ⇒ Object

Add or replace a cluster resource.



40
41
42
43
 
# File 'lib/async/grpc/xds/control_plane.rb', line 40

def update_cluster(name, resource = nil, **options)
	resource ||= ResourceBuilder.cluster(name, **options)
	update_resource(CLUSTER_TYPE, name.to_s, resource)
end

#update_endpoints(cluster_name, endpoints) ⇒ Object

Add or replace the endpoint assignment for a cluster.



48
49
50
51
52
53
54
 
# File 'lib/async/grpc/xds/control_plane.rb', line 48

def update_endpoints(cluster_name, endpoints)
	update_resource(
		ENDPOINT_TYPE,
		cluster_name.to_s,
		ResourceBuilder.cluster_load_assignment(cluster_name, endpoints)
	)
end

#update_resource(type_url, name, resource) ⇒ Object

Add or replace an xDS resource and notify subscribed streams.



72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/async/grpc/xds/control_plane.rb', line 72

def update_resource(type_url, name, resource)
	notify = false
	
	@mutex.synchronize do
		@resources[type_url][name] = resource
		@versions[type_url] += 1
		notify = true
	end
	
	notify_streams(type_url) if notify
end

#version(type_url) ⇒ Object

Get the current version for a resource type.



128
129
130
131
132
 
# File 'lib/async/grpc/xds/control_plane.rb', line 128

def version(type_url)
	@mutex.synchronize do
		@versions[type_url].to_s
	end
end