analytics: Add LoggingCount for messages read stats.

Whenever we use API queries to mark messages as read we now increment
two new LoggingCount stats, messages_read::hour and
messages_read_interactions::hour.

We add an early return in do_increment_logging_stat function if there
are no changes (increment is 0), as an optimization to avoid
unnecessary database queries.

We also log messages_read_interactions::hour Logging stat
as the number of API queries to mark messages as read.

We don't include tests for the case where do_update_pointer is called
because do_update_pointer will most likely be removed from the
codebase in the near future.

analytics/lib/counts.py

@@ -248,6 +248,9 @@ def do_aggregate_to_summary_table(stat: CountStat, end_time: datetime,
 def do_increment_logging_stat(zerver_object: Union[Realm, UserProfile, Stream], stat: CountStat,
                               subgroup: Optional[Union[str, int, bool]], event_time: datetime,
                               increment: int=1) -> None:
+    if not increment:
+        return
+
     table = stat.data_collector.output_table
     if table == RealmCount:
         id_args = {'realm': zerver_object}
@@ -645,6 +648,16 @@ def get_count_stats(realm: Optional[Realm]=None) -> Dict[str, CountStat]:
                   sql_data_collector(RealmCount, count_user_by_realm_query(realm), (UserProfile, 'is_bot')),
                   CountStat.DAY, interval=TIMEDELTA_MAX),
 
+        # Messages read stats.  messages_read::hour is the total
+        # number of messages read, whereas
+        # messages_read_interactions::hour tries to count the total
+        # number of UI interactions resulting in messages being marked
+        # as read (imperfect because of batching of some request
+        # types, but less likely to be overwhelmed by a single bulk
+        # operation).
+        LoggingCountStat('messages_read::hour', UserCount, CountStat.HOUR),
+        LoggingCountStat('messages_read_interactions::hour', UserCount, CountStat.HOUR),
+
         # User Activity stats
         # Stats that measure user activity in the UserActivityInterval sense.
 

analytics/tests/test_counts.py

@@ -38,9 +38,12 @@ from zerver.lib.actions import (
     do_create_user,
     do_deactivate_user,
     do_invite_users,
+    do_mark_all_as_read,
+    do_mark_stream_messages_as_read,
     do_reactivate_user,
     do_resend_user_invite_email,
     do_revoke_user_invite,
+    do_update_message_flags,
     update_user_activity_interval,
 )
 from zerver.lib.create_user import create_user
@@ -1151,6 +1154,39 @@ class TestLoggingCountStats(AnalyticsTestCase):
         do_resend_user_invite_email(PreregistrationUser.objects.first())
         assertInviteCountEquals(6)
 
+    def test_messages_read_hour(self) -> None:
+        read_count_property = 'messages_read::hour'
+        interactions_property = 'messages_read_interactions::hour'
+
+        user1 = self.create_user()
+        user2 = self.create_user()
+        stream, recipient = self.create_stream_with_recipient()
+        self.subscribe(user1, stream.name)
+        self.subscribe(user2, stream.name)
+
+        self.send_personal_message(user1, user2)
+        client = get_client("website")
+        do_mark_all_as_read(user2, client)
+        self.assertEqual(1, UserCount.objects.filter(property=read_count_property)
+                         .aggregate(Sum('value'))['value__sum'])
+        self.assertEqual(1, UserCount.objects.filter(property=interactions_property)
+                         .aggregate(Sum('value'))['value__sum'])
+
+        self.send_stream_message(user1, stream.name)
+        self.send_stream_message(user1, stream.name)
+        do_mark_stream_messages_as_read(user2, client, stream)
+        self.assertEqual(3, UserCount.objects.filter(property=read_count_property)
+                         .aggregate(Sum('value'))['value__sum'])
+        self.assertEqual(2, UserCount.objects.filter(property=interactions_property)
+                         .aggregate(Sum('value'))['value__sum'])
+
+        message = self.send_stream_message(user2, stream.name)
+        do_update_message_flags(user1, client, 'add', 'read', [message])
+        self.assertEqual(4, UserCount.objects.filter(property=read_count_property)
+                         .aggregate(Sum('value'))['value__sum'])
+        self.assertEqual(3, UserCount.objects.filter(property=interactions_property)
+                         .aggregate(Sum('value'))['value__sum'])
+
 class TestDeleteStats(AnalyticsTestCase):
     def test_do_drop_all_analytics_tables(self) -> None:
         user = self.create_user()

zerver/lib/actions.py

@@ -159,7 +159,7 @@ from zerver.lib.users import (
     get_api_key,
     user_profile_to_user_row,
 )
-from zerver.lib.utils import generate_api_key, log_statsd_event, statsd
+from zerver.lib.utils import generate_api_key, log_statsd_event
 from zerver.lib.validator import check_widget_content
 from zerver.lib.widget import do_widget_post_save_actions
 from zerver.models import (
@@ -4011,6 +4011,13 @@ def do_update_pointer(user_profile: UserProfile, client: Client,
                                    message__id__lte=pointer).extra(where=[UserMessage.where_unread()]) \
             .update(flags=F('flags').bitor(UserMessage.flags.read))
         do_clear_mobile_push_notifications_for_ids(user_profile, app_message_ids)
+        event_time = timezone_now()
+
+        count = len(app_message_ids)
+        do_increment_logging_stat(user_profile, COUNT_STATS['messages_read::hour'],
+                                  None, event_time, increment=count)
+        do_increment_logging_stat(user_profile, COUNT_STATS['messages_read_interactions::hour'],
+                                  None, event_time, increment=min(1, count))
 
     event = dict(type='pointer', pointer=pointer)
     send_event(user_profile.realm, event, [user_profile.id])
@@ -4066,9 +4073,14 @@ def do_mark_all_as_read(user_profile: UserProfile, client: Client) -> int:
         messages=[],  # we don't send messages, since the client reloads anyway
         all=True,
     )
+    event_time = timezone_now()
+
     send_event(user_profile.realm, event, [user_profile.id])
 
-    statsd.incr("mark_all_as_read", count)
+    do_increment_logging_stat(user_profile, COUNT_STATS['messages_read::hour'],
+                              None, event_time, increment=count)
+    do_increment_logging_stat(user_profile, COUNT_STATS['messages_read_interactions::hour'],
+                              None, event_time, increment=min(1, count))
 
     all_push_message_ids = UserMessage.objects.filter(
         user_profile=user_profile,
@@ -4115,10 +4127,15 @@ def do_mark_stream_messages_as_read(user_profile: UserProfile,
         messages=message_ids,
         all=False,
     )
+    event_time = timezone_now()
+
     send_event(user_profile.realm, event, [user_profile.id])
     do_clear_mobile_push_notifications_for_ids(user_profile, message_ids)
 
-    statsd.incr("mark_stream_as_read", count)
+    do_increment_logging_stat(user_profile, COUNT_STATS['messages_read::hour'],
+                              None, event_time, increment=count)
+    do_increment_logging_stat(user_profile, COUNT_STATS['messages_read_interactions::hour'],
+                              None, event_time, increment=min(1, count))
     return count
 
 def do_clear_mobile_push_notifications_for_ids(user_profile: UserProfile,
@@ -4202,9 +4219,13 @@ def do_update_message_flags(user_profile: UserProfile,
     send_event(user_profile.realm, event, [user_profile.id])
 
     if flag == "read" and operation == "add":
+        event_time = timezone_now()
         do_clear_mobile_push_notifications_for_ids(user_profile, messages)
 
-    statsd.incr(f"flags.{flag}.{operation}", count)
+        do_increment_logging_stat(user_profile, COUNT_STATS['messages_read::hour'],
+                                  None, event_time, increment=count)
+        do_increment_logging_stat(user_profile, COUNT_STATS['messages_read_interactions::hour'],
+                                  None, event_time, increment=min(1, count))
     return count
 
 class MessageUpdateUserInfoResult(TypedDict):