forked from googleapis/google-cloud-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcluster.py
More file actions
333 lines (260 loc) · 12.4 KB
/
cluster.py
File metadata and controls
333 lines (260 loc) · 12.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
# Copyright 2015 Google Inc. All rights reserved.
#
# 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.
"""User friendly container for Google Cloud Bigtable Cluster."""
import datetime
import re
from gcloud._helpers import _EPOCH
from gcloud.bigtable._generated import bigtable_cluster_data_pb2 as data_pb2
from gcloud.bigtable._generated import (
bigtable_cluster_service_messages_pb2 as messages_pb2)
from gcloud.bigtable.table import Table
_CLUSTER_NAME_RE = re.compile(r'^projects/(?P<project>[^/]+)/'
r'zones/(?P<zone>[^/]+)/clusters/'
r'(?P<cluster_id>[a-z][-a-z0-9]*)$')
_OPERATION_NAME_RE = re.compile(r'^operations/projects/([^/]+)/zones/([^/]+)/'
r'clusters/([a-z][-a-z0-9]*)/operations/'
r'(?P<operation_id>\d+)$')
_DEFAULT_SERVE_NODES = 3
_TYPE_URL_BASE = 'type.googleapis.com/google.bigtable.'
_ADMIN_TYPE_URL_BASE = _TYPE_URL_BASE + 'admin.cluster.v1.'
_CLUSTER_CREATE_METADATA = _ADMIN_TYPE_URL_BASE + 'CreateClusterMetadata'
_TYPE_URL_MAP = {
_CLUSTER_CREATE_METADATA: messages_pb2.CreateClusterMetadata,
}
def _get_pb_property_value(message_pb, property_name):
"""Return a message field value.
:type message_pb: :class:`google.protobuf.message.Message`
:param message_pb: The message to check for ``property_name``.
:type property_name: str
:param property_name: The property value to check against.
:rtype: object
:returns: The value of ``property_name`` set on ``message_pb``.
:raises: :class:`ValueError <exceptions.ValueError>` if the result returned
from the ``message_pb`` does not contain the ``property_name``
value.
"""
# Make sure `property_name` is set on the response.
# NOTE: As of proto3, HasField() only works for message fields, not for
# singular (non-message) fields.
all_fields = set([field.name for field in message_pb._fields])
if property_name not in all_fields:
raise ValueError('Message does not contain %s.' % (property_name,))
return getattr(message_pb, property_name)
def _prepare_create_request(cluster):
"""Creates a protobuf request for a CreateCluster request.
:type cluster: :class:`Cluster`
:param cluster: The cluster to be created.
:rtype: :class:`.messages_pb2.CreateClusterRequest`
:returns: The CreateCluster request object containing the cluster info.
"""
zone_full_name = ('projects/' + cluster._client.project +
'/zones/' + cluster.zone)
return messages_pb2.CreateClusterRequest(
name=zone_full_name,
cluster_id=cluster.cluster_id,
cluster=data_pb2.Cluster(
display_name=cluster.display_name,
serve_nodes=cluster.serve_nodes,
),
)
def _pb_timestamp_to_datetime(timestamp):
"""Convert a Timestamp protobuf to a datetime object.
:type timestamp: :class:`._generated.timestamp_pb2.Timestamp`
:param timestamp: A Google returned timestamp protobuf.
:rtype: :class:`datetime.datetime`
:returns: A UTC datetime object converted from a protobuf timestamp.
"""
return (
_EPOCH +
datetime.timedelta(
seconds=timestamp.seconds,
microseconds=(timestamp.nanos / 1000.0),
)
)
def _parse_pb_any_to_native(any_val, expected_type=None):
"""Convert a serialized "google.protobuf.Any" value to actual type.
:type any_val: :class:`gcloud.bigtable._generated.any_pb2.Any`
:param any_val: A serialized protobuf value container.
:type expected_type: str
:param expected_type: (Optional) The type URL we expect ``any_val``
to have.
:rtype: object
:returns: The de-serialized object.
:raises: :class:`ValueError <exceptions.ValueError>` if the
``expected_type`` does not match the ``type_url`` on the input.
"""
if expected_type is not None and expected_type != any_val.type_url:
raise ValueError('Expected type: %s, Received: %s' % (
expected_type, any_val.type_url))
container_class = _TYPE_URL_MAP[any_val.type_url]
return container_class.FromString(any_val.value)
def _process_operation(operation_pb):
"""Processes a create protobuf response.
:type operation_pb: :class:`operations_pb2.Operation`
:param operation_pb: The long-running operation response from a
Create/Update/Undelete cluster request.
:rtype: tuple
:returns: A pair of an integer and datetime stamp. The integer is the ID
of the operation (``operation_id``) and the timestamp when
the create operation began (``operation_begin``).
:raises: :class:`ValueError <exceptions.ValueError>` if the operation name
doesn't match the :data:`_OPERATION_NAME_RE` regex.
"""
match = _OPERATION_NAME_RE.match(operation_pb.name)
if match is None:
raise ValueError('Operation name was not in the expected '
'format after a cluster modification.',
operation_pb.name)
operation_id = int(match.group('operation_id'))
request_metadata = _parse_pb_any_to_native(operation_pb.metadata)
operation_begin = _pb_timestamp_to_datetime(
request_metadata.request_time)
return operation_id, operation_begin
class Cluster(object):
"""Representation of a Google Cloud Bigtable Cluster.
:type zone: str
:param zone: The name of the zone where the cluster resides.
:type cluster_id: str
:param cluster_id: The ID of the cluster.
:type client: :class:`.client.Client`
:param client: The client that owns the cluster. Provides
authorization and a project ID.
:type display_name: str
:param display_name: (Optional) The display name for the cluster in the
Cloud Console UI. (Must be between 4 and 30
characters.) If this value is not set in the
constructor, will fall back to the cluster ID.
:type serve_nodes: int
:param serve_nodes: (Optional) The number of nodes in the cluster.
Defaults to 3.
"""
def __init__(self, zone, cluster_id, client,
display_name=None, serve_nodes=_DEFAULT_SERVE_NODES):
self.zone = zone
self.cluster_id = cluster_id
self.display_name = display_name or cluster_id
self.serve_nodes = serve_nodes
self._client = client
self._operation_type = None
self._operation_id = None
self._operation_begin = None
def table(self, table_id):
"""Factory to create a table associated with this cluster.
:type table_id: str
:param table_id: The ID of the table.
:rtype: :class:`Table <gcloud.bigtable.table.Table>`
:returns: The table owned by this cluster.
"""
return Table(table_id, self)
def _update_from_pb(self, cluster_pb):
self.display_name = _get_pb_property_value(cluster_pb, 'display_name')
self.serve_nodes = _get_pb_property_value(cluster_pb, 'serve_nodes')
@classmethod
def from_pb(cls, cluster_pb, client):
"""Creates a cluster instance from a protobuf.
:type cluster_pb: :class:`bigtable_cluster_data_pb2.Cluster`
:param cluster_pb: A cluster protobuf object.
:type client: :class:`.client.Client`
:param client: The client that owns the cluster.
:rtype: :class:`Cluster`
:returns: The cluster parsed from the protobuf response.
:raises: :class:`ValueError <exceptions.ValueError>` if the cluster
name does not match :data:`_CLUSTER_NAME_RE` or if the parsed
project ID does not match the project ID on the client.
"""
match = _CLUSTER_NAME_RE.match(cluster_pb.name)
if match is None:
raise ValueError('Cluster protobuf name was not in the '
'expected format.', cluster_pb.name)
if match.group('project') != client.project:
raise ValueError('Project ID on cluster does not match the '
'project ID on the client')
result = cls(match.group('zone'), match.group('cluster_id'), client)
result._update_from_pb(cluster_pb)
return result
@property
def name(self):
"""Cluster name used in requests.
.. note::
This property will not change if ``zone`` and ``cluster_id`` do not,
but the return value is not cached.
The cluster name is of the form
``"projects/{project}/zones/{zone}/clusters/{cluster_id}"``
:rtype: str
:returns: The cluster name.
"""
return (self._client.project_name + '/zones/' + self.zone +
'/clusters/' + self.cluster_id)
def __eq__(self, other):
if not isinstance(other, self.__class__):
return False
# NOTE: This does not compare the configuration values, such as
# the serve_nodes or display_name. Instead, it only compares
# identifying values zone, cluster ID and client. This is
# intentional, since the same cluster can be in different states
# if not synchronized. Clusters with similar zone/cluster
# settings but different clients can't be used in the same way.
return (other.zone == self.zone and
other.cluster_id == self.cluster_id and
other._client == self._client)
def __ne__(self, other):
return not self.__eq__(other)
def reload(self):
"""Reload the metadata for this cluster."""
request_pb = messages_pb2.GetClusterRequest(name=self.name)
# We expect a `._generated.bigtable_cluster_data_pb2.Cluster`.
cluster_pb = self._client._cluster_stub.GetCluster(
request_pb, self._client.timeout_seconds)
# NOTE: _update_from_pb does not check that the project, zone and
# cluster ID on the response match the request.
self._update_from_pb(cluster_pb)
def create(self):
"""Create this cluster.
.. note::
Uses the ``project``, ``zone`` and ``cluster_id`` on the current
:class:`Cluster` in addition to the ``display_name`` and
``serve_nodes``. If you'd like to change them before creating,
reset the values via
.. code:: python
cluster.display_name = 'New display name'
cluster.cluster_id = 'i-changed-my-mind'
before calling :meth:`create`.
"""
request_pb = _prepare_create_request(self)
# We expect an `operations_pb2.Operation`.
cluster_pb = self._client._cluster_stub.CreateCluster(
request_pb, self._client.timeout_seconds)
self._operation_type = 'create'
self._operation_id, self._operation_begin = _process_operation(
cluster_pb.current_operation)
def delete(self):
"""Delete this cluster.
Marks a cluster and all of its tables for permanent deletion in 7 days.
Immediately upon completion of the request:
* Billing will cease for all of the cluster's reserved resources.
* The cluster's ``delete_time`` field will be set 7 days in the future.
Soon afterward:
* All tables within the cluster will become unavailable.
Prior to the cluster's ``delete_time``:
* The cluster can be recovered with a call to ``UndeleteCluster``.
* All other attempts to modify or delete the cluster will be rejected.
At the cluster's ``delete_time``:
* The cluster and **all of its tables** will immediately and
irrevocably disappear from the API, and their data will be
permanently deleted.
"""
request_pb = messages_pb2.DeleteClusterRequest(name=self.name)
# We expect a `._generated.empty_pb2.Empty`
self._client._cluster_stub.DeleteCluster(
request_pb, self._client.timeout_seconds)