-
Notifications
You must be signed in to change notification settings - Fork 52
/
_gapic.py
643 lines (538 loc) · 23.4 KB
/
_gapic.py
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
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
# Copyright 2016 Google LLC
#
# 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.
"""Wrapper for adapting the autogenerated gapic client to the hand-written
client."""
from google.cloud.logging_v2.services.config_service_v2 import ConfigServiceV2Client
from google.cloud.logging_v2.services.logging_service_v2 import LoggingServiceV2Client
from google.cloud.logging_v2.services.metrics_service_v2 import MetricsServiceV2Client
from google.cloud.logging_v2.types import CreateSinkRequest
from google.cloud.logging_v2.types import UpdateSinkRequest
from google.cloud.logging_v2.types import ListSinksRequest
from google.cloud.logging_v2.types import ListLogMetricsRequest
from google.cloud.logging_v2.types import ListLogEntriesRequest
from google.cloud.logging_v2.types import WriteLogEntriesRequest
from google.cloud.logging_v2.types import LogSink
from google.cloud.logging_v2.types import LogMetric
from google.cloud.logging_v2.types import LogEntry as LogEntryPB
from google.protobuf.json_format import MessageToDict
from google.protobuf.json_format import ParseDict
from google.cloud.logging_v2._helpers import entry_from_resource
from google.cloud.logging_v2.sink import Sink
from google.cloud.logging_v2.metric import Metric
from google.api_core import client_info
from google.api_core import gapic_v1
class _LoggingAPI(object):
"""Helper mapping logging-related APIs."""
def __init__(self, gapic_api, client):
self._gapic_api = gapic_api
self._client = client
def list_entries(
self,
resource_names,
*,
filter_=None,
order_by=None,
max_results=None,
page_size=None,
page_token=None,
):
"""Return a generator of log entry resources.
Args:
resource_names (Sequence[str]): Names of one or more parent resources
from which to retrieve log entries:
::
"projects/[PROJECT_ID]"
"organizations/[ORGANIZATION_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]"
"folders/[FOLDER_ID]"
filter_ (str): a filter expression. See
https://cloud.google.com/logging/docs/view/advanced_filters
order_by (str) One of :data:`~logging_v2.ASCENDING`
or :data:`~logging_v2.DESCENDING`.
max_results (Optional[int]):
Optional. The maximum number of entries to return.
Non-positive values are treated as 0. If None, uses API defaults.
page_size (int): number of entries to fetch in each API call. Although
requests are paged internally, logs are returned by the generator
one at a time. If not passed, defaults to a value set by the API.
page_token (str): opaque marker for the starting "page" of entries. If not
passed, the API will return the first page of entries.
Returns:
Generator[~logging_v2.LogEntry]
"""
# full resource names are expected by the API
resource_names = resource_names
request = ListLogEntriesRequest(
resource_names=resource_names,
filter=filter_,
order_by=order_by,
page_size=page_size,
page_token=page_token,
)
response = self._gapic_api.list_log_entries(request=request)
log_iter = iter(response)
# We attach a mutable loggers dictionary so that as Logger
# objects are created by entry_from_resource, they can be
# re-used by other log entries from the same logger.
loggers = {}
if max_results is not None and max_results < 0:
raise ValueError("max_results must be positive")
# create generator
def log_entries_pager(log_iter):
i = 0
for entry in log_iter:
if max_results is not None and i >= max_results:
break
log_entry_dict = _parse_log_entry(LogEntryPB.pb(entry))
yield entry_from_resource(log_entry_dict, self._client, loggers=loggers)
i += 1
return log_entries_pager(log_iter)
def write_entries(
self,
entries,
*,
logger_name=None,
resource=None,
labels=None,
partial_success=True,
dry_run=False,
):
"""Log an entry resource via a POST request
Args:
entries (Sequence[Mapping[str, ...]]): sequence of mappings representing
the log entry resources to log.
logger_name (Optional[str]): name of default logger to which to log the entries;
individual entries may override.
resource(Optional[Mapping[str, ...]]): default resource to associate with entries;
individual entries may override.
labels (Optional[Mapping[str, ...]]): default labels to associate with entries;
individual entries may override.
partial_success (Optional[bool]): Whether valid entries should be written even if
some other entries fail due to INVALID_ARGUMENT or
PERMISSION_DENIED errors. If any entry is not written, then
the response status is the error associated with one of the
failed entries and the response includes error details keyed
by the entries' zero-based index in the ``entries.write``
method.
dry_run (Optional[bool]):
If true, the request should expect normal response,
but the entries won't be persisted nor exported.
Useful for checking whether the logging API endpoints are working
properly before sending valuable data.
"""
log_entry_pbs = [_log_entry_mapping_to_pb(entry) for entry in entries]
request = WriteLogEntriesRequest(
log_name=logger_name,
resource=resource,
labels=labels,
entries=log_entry_pbs,
partial_success=partial_success,
)
self._gapic_api.write_log_entries(request=request)
def logger_delete(self, logger_name):
"""Delete all entries in a logger.
Args:
logger_name (str): The resource name of the log to delete:
::
"projects/[PROJECT_ID]/logs/[LOG_ID]"
"organizations/[ORGANIZATION_ID]/logs/[LOG_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]/logs/[LOG_ID]"
"folders/[FOLDER_ID]/logs/[LOG_ID]"
``[LOG_ID]`` must be URL-encoded. For example,
``"projects/my-project-id/logs/syslog"``,
``"organizations/1234567890/logs/cloudresourcemanager.googleapis.com%2Factivity"``.
"""
self._gapic_api.delete_log(log_name=logger_name)
class _SinksAPI(object):
"""Helper mapping sink-related APIs."""
def __init__(self, gapic_api, client):
self._gapic_api = gapic_api
self._client = client
def list_sinks(self, parent, *, max_results=None, page_size=None, page_token=None):
"""List sinks for the parent resource.
Args:
parent (str): The parent resource whose sinks are to be listed:
::
"projects/[PROJECT_ID]"
"organizations/[ORGANIZATION_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]"
"folders/[FOLDER_ID]".
max_results (Optional[int]):
Optional. The maximum number of entries to return.
Non-positive values are treated as 0. If None, uses API defaults.
page_size (int): number of entries to fetch in each API call. Although
requests are paged internally, logs are returned by the generator
one at a time. If not passed, defaults to a value set by the API.
page_token (str): opaque marker for the starting "page" of entries. If not
passed, the API will return the first page of entries.
Returns:
Generator[~logging_v2.Sink]
"""
request = ListSinksRequest(
parent=parent, page_size=page_size, page_token=page_token
)
response = self._gapic_api.list_sinks(request)
sink_iter = iter(response)
if max_results is not None and max_results < 0:
raise ValueError("max_results must be positive")
def sinks_pager(sink_iter):
i = 0
for entry in sink_iter:
if max_results is not None and i >= max_results:
break
# Convert the GAPIC sink type into the handwritten `Sink` type
yield Sink.from_api_repr(LogSink.to_dict(entry), client=self._client)
i += 1
return sinks_pager(sink_iter)
def sink_create(
self, parent, sink_name, filter_, destination, *, unique_writer_identity=False
):
"""Create a sink resource.
See
https://cloud.google.com/logging/docs/reference/v2/rest/v2/projects.sinks/create
Args:
parent(str): The resource in which to create the sink,
including the parent resource and the sink identifier:
::
"projects/[PROJECT_ID]"
"organizations/[ORGANIZATION_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]"
"folders/[FOLDER_ID]".
sink_name (str): The name of the sink.
filter_ (str): The advanced logs filter expression defining the
entries exported by the sink.
destination (str): Destination URI for the entries exported by
the sink.
unique_writer_identity (Optional[bool]): determines the kind of
IAM identity returned as writer_identity in the new sink.
Returns:
dict: The sink resource returned from the API (converted from a
protobuf to a dictionary).
"""
sink_pb = LogSink(name=sink_name, filter=filter_, destination=destination)
request = CreateSinkRequest(
parent=parent, sink=sink_pb, unique_writer_identity=unique_writer_identity
)
created_pb = self._gapic_api.create_sink(request=request)
return MessageToDict(
LogSink.pb(created_pb),
preserving_proto_field_name=False,
)
def sink_get(self, sink_name):
"""Retrieve a sink resource.
Args:
sink_name (str): The resource name of the sink,
including the parent resource and the sink identifier:
::
"projects/[PROJECT_ID]/sinks/[SINK_ID]"
"organizations/[ORGANIZATION_ID]/sinks/[SINK_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]/sinks/[SINK_ID]"
"folders/[FOLDER_ID]/sinks/[SINK_ID]"
Returns:
dict: The sink object returned from the API (converted from a
protobuf to a dictionary).
"""
sink_pb = self._gapic_api.get_sink(sink_name=sink_name)
# NOTE: LogSink message type does not have an ``Any`` field
# so `MessageToDict`` can safely be used.
return MessageToDict(
LogSink.pb(sink_pb),
preserving_proto_field_name=False,
)
def sink_update(
self,
sink_name,
filter_,
destination,
*,
unique_writer_identity=False,
):
"""Update a sink resource.
Args:
sink_name (str): Required. The resource name of the sink,
including the parent resource and the sink identifier:
::
"projects/[PROJECT_ID]/sinks/[SINK_ID]"
"organizations/[ORGANIZATION_ID]/sinks/[SINK_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]/sinks/[SINK_ID]"
"folders/[FOLDER_ID]/sinks/[SINK_ID]"
filter_ (str): The advanced logs filter expression defining the
entries exported by the sink.
destination (str): destination URI for the entries exported by
the sink.
unique_writer_identity (Optional[bool]): determines the kind of
IAM identity returned as writer_identity in the new sink.
Returns:
dict: The sink resource returned from the API (converted from a
protobuf to a dictionary).
"""
name = sink_name.split("/")[-1] # parse name out of full resoure name
sink_pb = LogSink(
name=name,
filter=filter_,
destination=destination,
)
request = UpdateSinkRequest(
sink_name=sink_name,
sink=sink_pb,
unique_writer_identity=unique_writer_identity,
)
sink_pb = self._gapic_api.update_sink(request=request)
# NOTE: LogSink message type does not have an ``Any`` field
# so `MessageToDict`` can safely be used.
return MessageToDict(
LogSink.pb(sink_pb),
preserving_proto_field_name=False,
)
def sink_delete(self, sink_name):
"""Delete a sink resource.
Args:
sink_name (str): Required. The full resource name of the sink to delete,
including the parent resource and the sink identifier:
::
"projects/[PROJECT_ID]/sinks/[SINK_ID]"
"organizations/[ORGANIZATION_ID]/sinks/[SINK_ID]"
"billingAccounts/[BILLING_ACCOUNT_ID]/sinks/[SINK_ID]"
"folders/[FOLDER_ID]/sinks/[SINK_ID]"
Example: ``"projects/my-project-id/sinks/my-sink-id"``.
"""
self._gapic_api.delete_sink(sink_name=sink_name)
class _MetricsAPI(object):
"""Helper mapping sink-related APIs."""
def __init__(self, gapic_api, client):
self._gapic_api = gapic_api
self._client = client
def list_metrics(
self, project, *, max_results=None, page_size=None, page_token=None
):
"""List metrics for the project associated with this client.
Args:
project (str): ID of the project whose metrics are to be listed.
max_results (Optional[int]):
Optional. The maximum number of entries to return.
Non-positive values are treated as 0. If None, uses API defaults.
page_size (int): number of entries to fetch in each API call. Although
requests are paged internally, logs are returned by the generator
one at a time. If not passed, defaults to a value set by the API.
page_token (str): opaque marker for the starting "page" of entries. If not
passed, the API will return the first page of entries.
Returns:
Generator[logging_v2.Metric]
"""
path = f"projects/{project}"
request = ListLogMetricsRequest(
parent=path,
page_size=page_size,
page_token=page_token,
)
response = self._gapic_api.list_log_metrics(request=request)
metric_iter = iter(response)
if max_results is not None and max_results < 0:
raise ValueError("max_results must be positive")
def metrics_pager(metric_iter):
i = 0
for entry in metric_iter:
if max_results is not None and i >= max_results:
break
# Convert GAPIC metrics type into handwritten `Metric` type
yield Metric.from_api_repr(
LogMetric.to_dict(entry), client=self._client
)
i += 1
return metrics_pager(metric_iter)
def metric_create(self, project, metric_name, filter_, description):
"""Create a metric resource.
See
https://cloud.google.com/logging/docs/reference/v2/rest/v2/projects.metrics/create
Args:
project (str): ID of the project in which to create the metric.
metric_name (str): The name of the metric
filter_ (str): The advanced logs filter expression defining the
entries exported by the metric.
description (str): description of the metric.
"""
parent = f"projects/{project}"
metric_pb = LogMetric(name=metric_name, filter=filter_, description=description)
self._gapic_api.create_log_metric(parent=parent, metric=metric_pb)
def metric_get(self, project, metric_name):
"""Retrieve a metric resource.
Args:
project (str): ID of the project containing the metric.
metric_name (str): The name of the metric
Returns:
dict: The metric object returned from the API (converted from a
protobuf to a dictionary).
"""
path = f"projects/{project}/metrics/{metric_name}"
metric_pb = self._gapic_api.get_log_metric(metric_name=path)
# NOTE: LogMetric message type does not have an ``Any`` field
# so `MessageToDict`` can safely be used.
return MessageToDict(
LogMetric.pb(metric_pb),
preserving_proto_field_name=False,
)
def metric_update(
self,
project,
metric_name,
filter_,
description,
):
"""Update a metric resource.
Args:
project (str): ID of the project containing the metric.
metric_name (str): the name of the metric
filter_ (str): the advanced logs filter expression defining the
entries exported by the metric.
description (str): description of the metric.
Returns:
The metric object returned from the API (converted from a
protobuf to a dictionary).
"""
path = f"projects/{project}/metrics/{metric_name}"
metric_pb = LogMetric(
name=path,
filter=filter_,
description=description,
)
metric_pb = self._gapic_api.update_log_metric(
metric_name=path, metric=metric_pb
)
# NOTE: LogMetric message type does not have an ``Any`` field
# so `MessageToDict`` can safely be used.
return MessageToDict(
LogMetric.pb(metric_pb),
preserving_proto_field_name=False,
)
def metric_delete(self, project, metric_name):
"""Delete a metric resource.
Args:
project (str): ID of the project containing the metric.
metric_name (str): The name of the metric
"""
path = f"projects/{project}/metrics/{metric_name}"
self._gapic_api.delete_log_metric(metric_name=path)
def _parse_log_entry(entry_pb):
"""Special helper to parse ``LogEntry`` protobuf into a dictionary.
The ``proto_payload`` field in ``LogEntry`` is of type ``Any``. This
can be problematic if the type URL in the payload isn't in the
``google.protobuf`` registry. To help with parsing unregistered types,
this function will remove ``proto_payload`` before parsing.
Args:
entry_pb (LogEntry): Log entry protobuf.
Returns:
dict: The parsed log entry. The ``protoPayload`` key may contain
the raw ``Any`` protobuf from ``entry_pb.proto_payload`` if
it could not be parsed.
"""
try:
return MessageToDict(
entry_pb,
preserving_proto_field_name=False,
)
except TypeError:
if entry_pb.HasField("proto_payload"):
proto_payload = entry_pb.proto_payload
entry_pb.ClearField("proto_payload")
entry_mapping = MessageToDict(
entry_pb,
preserving_proto_field_name=False,
)
entry_mapping["protoPayload"] = proto_payload
return entry_mapping
else:
raise
def _log_entry_mapping_to_pb(mapping):
"""Helper for :meth:`write_entries`, et aliae
Performs "impedance matching" between the protobuf attrs and
the keys expected in the JSON API.
"""
entry_pb = LogEntryPB.pb(LogEntryPB())
# NOTE: We assume ``mapping`` was created in ``Batch.commit``
# or ``Logger._make_entry_resource``. In either case, if
# the ``protoPayload`` key is present, we assume that the
# type URL is registered with ``google.protobuf`` and will
# not cause any issues in the JSON->protobuf conversion
# of the corresponding ``proto_payload`` in the log entry
# (it is an ``Any`` field).
ParseDict(mapping, entry_pb)
return LogEntryPB(entry_pb)
def _client_info_to_gapic(input_info):
"""
Helper function to convert api_core.client_info to
api_core.gapic_v1.client_info subclass
"""
return gapic_v1.client_info.ClientInfo(
python_version=input_info.python_version,
grpc_version=input_info.grpc_version,
api_core_version=input_info.api_core_version,
gapic_version=input_info.gapic_version,
client_library_version=input_info.client_library_version,
user_agent=input_info.user_agent,
rest_version=input_info.rest_version,
)
def make_logging_api(client):
"""Create an instance of the Logging API adapter.
Args:
client (~logging_v2.client.Client): The client
that holds configuration details.
Returns:
_LoggingAPI: A metrics API instance with the proper credentials.
"""
info = client._client_info
if isinstance(info, client_info.ClientInfo):
# convert into gapic-compatible subclass
info = _client_info_to_gapic(info)
generated = LoggingServiceV2Client(
credentials=client._credentials,
client_info=info,
client_options=client._client_options,
)
return _LoggingAPI(generated, client)
def make_metrics_api(client):
"""Create an instance of the Metrics API adapter.
Args:
client (~logging_v2.client.Client): The client
that holds configuration details.
Returns:
_MetricsAPI: A metrics API instance with the proper credentials.
"""
info = client._client_info
if isinstance(info, client_info.ClientInfo):
# convert into gapic-compatible subclass
info = _client_info_to_gapic(info)
generated = MetricsServiceV2Client(
credentials=client._credentials,
client_info=info,
client_options=client._client_options,
)
return _MetricsAPI(generated, client)
def make_sinks_api(client):
"""Create an instance of the Sinks API adapter.
Args:
client (~logging_v2.client.Client): The client
that holds configuration details.
Returns:
_SinksAPI: A metrics API instance with the proper credentials.
"""
info = client._client_info
if isinstance(info, client_info.ClientInfo):
# convert into gapic-compatible subclass
info = _client_info_to_gapic(info)
generated = ConfigServiceV2Client(
credentials=client._credentials,
client_info=info,
client_options=client._client_options,
)
return _SinksAPI(generated, client)