ajamato commented on a change in pull request #15328:
URL: https://github.com/apache/beam/pull/15328#discussion_r689793637
##########
File path: sdks/python/apache_beam/io/gcp/datastore/v1new/datastoreio.py
##########
@@ -265,10 +272,33 @@ def get_estimated_num_splits(client, query):
class _QueryFn(DoFn):
"""A DoFn that fetches entities from Cloud Datastore, for a given query."""
def process(self, query, *unused_args, **unused_kwargs):
+ if query.namespace is None:
+ query.namespace = ''
_client = helper.get_client(query.project, query.namespace)
client_query = query._to_client_query(_client)
- for client_entity in client_query.fetch(query.limit):
- yield types.Entity.from_client_entity(client_entity)
+ # Create request count metric
+ resource = resource_identifiers.DatastoreNamespace(
+ query.project, query.namespace)
+ labels = {
+ monitoring_infos.SERVICE_LABEL: 'Datastore',
+ monitoring_infos.METHOD_LABEL: 'BatchDatastoreRead',
+ monitoring_infos.RESOURCE_LABEL: resource,
+ monitoring_infos.DATASTORE_NAMESPACE_LABEL: query.namespace,
+ monitoring_infos.DATASTORE_PROJECT_ID_LABEL: query.project,
+ monitoring_infos.STATUS_LABEL: 'ok'
+ }
+ service_call_metric = ServiceCallMetric(
+ request_count_urn=monitoring_infos.API_REQUEST_COUNT_URN,
+ base_labels=labels)
+ try:
+ for client_entity in client_query.fetch(query.limit):
+ yield types.Entity.from_client_entity(client_entity)
+ service_call_metric.call('ok')
+ except (ClientError, GoogleAPICallError) as e:
Review comment:
I don't see any imports for ClientError, GoogleAPICallError
Did you confirm that this client library raises this exception type? with
your logging?
Please remove this code if not.
##########
File path: sdks/python/apache_beam/io/gcp/datastore/v1new/datastoreio.py
##########
@@ -26,20 +26,27 @@
import logging
import time
+from google.api_core.exceptions import ClientError
Review comment:
Please import this in a try block as shown here
https://github.com/apache/beam/blob/39cf3fccbb3950b9fd0c8938301d192c3ca85cce/sdks/python/apache_beam/io/gcp/bigquery_tools.py#L68
##########
File path: sdks/python/apache_beam/io/gcp/datastore/v1new/datastoreio_test.py
##########
@@ -177,6 +180,50 @@ def test_write_mutations_non_retryable_error(self):
mock_throttler, rpc_stats_callback, throttle_delay=0)
rpc_stats_callback.assert_called_once_with(errors=1)
+ def test_write_mutations_metric_on_failure(self):
Review comment:
Please write a similar test for the read side:
invoking _QueryFn
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
