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 = 
Cluster::TYPE_URL
ENDPOINT_TYPE = 
Endpoint::TYPE_URL

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

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

Initialize an empty control plane.



29
30
31
32
33
34
35
 
# File 'lib/async/grpc/xds/control_plane.rb', line 29

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.



37
38
39
 
# File 'lib/async/grpc/xds/control_plane.rb', line 37

def identifier
  @identifier
end

Instance Method Details

#register_stream(stream) ⇒ Object

Register a stream to receive resource-change notifications.



156
157
158
159
160
 
# File 'lib/async/grpc/xds/control_plane.rb', line 156

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

#remove_cluster(name) ⇒ Object

Remove a cluster resource.



61
62
63
 
# File 'lib/async/grpc/xds/control_plane.rb', line 61

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

#remove_endpoints(cluster_name) ⇒ Object

Remove the endpoint assignment for a cluster.



67
68
69
 
# File 'lib/async/grpc/xds/control_plane.rb', line 67

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.



90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/async/grpc/xds/control_plane.rb', line 90

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.



164
165
166
167
168
 
# File 'lib/async/grpc/xds/control_plane.rb', line 164

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.



106
107
108
109
110
 
# File 'lib/async/grpc/xds/control_plane.rb', line 106

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.



116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/async/grpc/xds/control_plane.rb', line 116

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.



141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/async/grpc/xds/control_plane.rb', line 141

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| Google::Protobuf::Any.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.



43
44
45
46
 
# File 'lib/async/grpc/xds/control_plane.rb', line 43

def update_cluster(name, resource = nil, **options)
	resource ||= Cluster.build(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.



51
52
53
54
55
56
57
 
# File 'lib/async/grpc/xds/control_plane.rb', line 51

def update_endpoints(cluster_name, endpoints)
	update_resource(
		ENDPOINT_TYPE,
		cluster_name.to_s,
		Endpoint.build(cluster_name, endpoints)
	)
end

#update_resource(type_url, name, resource) ⇒ Object

Add or replace an xDS resource and notify subscribed streams.



75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/async/grpc/xds/control_plane.rb', line 75

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.



131
132
133
134
135
 
# File 'lib/async/grpc/xds/control_plane.rb', line 131

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