From 94b38c14676d460acd840867b4bd66eccdbf9986 Mon Sep 17 00:00:00 2001 From: yuenmichelle1 Date: Tue, 19 Dec 2023 10:47:24 -0600 Subject: [PATCH 1/4] update kinesis stream. to log duplicates into sentry --- app/models/kinesis_stream.rb | 78 +++++++++++++++++++++++++++++++++--- 1 file changed, 72 insertions(+), 6 deletions(-) diff --git a/app/models/kinesis_stream.rb b/app/models/kinesis_stream.rb index 563e55b..ad2b228 100644 --- a/app/models/kinesis_stream.rb +++ b/app/models/kinesis_stream.rb @@ -9,14 +9,80 @@ def initialize def create_events(payload) receive(payload) - # TO DO (possibly?): We may want to consider doing these upserts/inserts in batches to improve performance. - CommentEvent.upsert_all(@comment_events, unique_by: %i[comment_id event_time]) unless @comment_events.empty? - unless @classification_events.empty? - ClassificationEvent.upsert_all(@classification_events, - unique_by: %i[classification_id event_time]) + + # Because ERAS is one of the ONLY receiving apps that receives from kinesis and BULK UPSERTS (feature of Rails 6+), it has caught duplicates on payload from kinesis stream + # See: https://zooniverse-27.sentry.io/issues/4717869260/?project=4506117954011141&query=is%3Aunresolved&referrer=issue-stream&statsPeriod=14d&stream_index=3 + # EVEN THOUGH de-duping the payload by id before upserting should resolve issues with ERAS, (since ERAS only cares about counting the classification/comment once), + # UNFORTUNATELY, there are other apps (eg. Caesar, Tove) that rely on the kinesis stream and where duplicates in payload may affect results. + # Since ERAS is one of the only places we can catch this error (because of how it can bulk upsert), the team has decided to log the error to Sentry when the duplicates in payload occurs + # and also log the payload to Sentry. + ## Should note that this duplicate error has been seen before: + ## SEE: https://github.com/zooniverse/zoo-stats-api-graphql/pull/128 + ## ALSO NOTING: THIS CATCH, LOG, DEDUPE AND TRY UPSERTING AGAIN TO DB SITUATION IS TEMPORARY AND ONLY USED + ## TO SEE WHAT THE DUPES IN THE KINESIS PAYLOAD ARE. + ## ONE MORE NOTE: per Kinesis docs, it is VERY possible for Kinesis stream to send duplicates and + ## the recommendation of AWS is to appropriately handle process records. + + upsert_comments unless @comment_events.empty? + upsert_classifications unless @classification_events.empty? + upsert_classification_user_groups unless @classification_user_groups.empty? + end + + def upsert_comments + CommentEvent.upsert_all(@comment_events, unique_by: %i[comment_id event_time]) + rescue StandardError => e + crumb = Sentry::Breadcrumb.new( + category: 'upsert_error_in_comments', + message: 'Comment Events Upsert Error', + data: { + payload: @comment_events, + error_message: e.message + }, + level: 'warning' + ) + Sentry.add_breadcrumb(crumb) + if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') + @comment_events = @comment_events.uniq { |comment| comment[:comment_id] } + retry end + end - ClassificationUserGroup.upsert_all(@classification_user_groups.flatten, unique_by: %i[classification_id event_time user_group_id user_id]) unless @classification_user_groups.empty? + def upsert_classifications + ClassificationEvent.upsert_all(@classification_events, unique_by: %i[classification_id event_time]) + rescue StandardError => e + crumb = Sentry::Breadcrumb.new( + category: 'upsert_error_in_classifications', + message: 'Classification Events Upsert Error', + data: { + payload: @classification_events, + error_message: e.message + }, + level: 'warning' + ) + Sentry.add_breadcrumb(crumb) + if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') + @classification_events = @classification_events.uniq { |classification| classification[:classification_id] } + retry + end + end + + def upsert_classification_user_groups + ClassificationUserGroup.upsert_all(@classification_user_groups.flatten, unique_by: %i[classification_id event_time user_group_id user_id]) + rescue StandardError => e + crumb = Sentry::Breadcrumb.new( + category: 'upsert_error_in_classifications_user_groups', + message: 'Classification User Groups Upsert Error', + data: { + payload: @classification_user_groups, + error_message: e.message + }, + level: 'warning' + ) + Sentry.add_breadcrumb(crumb) + if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') + @classification_user_groups = @classification_user_groups.uniq { |cug| [cug[:classification_id], cug[:user_group_id]] } + retry + end end def receive(payload) From a0a5de97468ebc75284ae261cd84f79796b9b8da Mon Sep 17 00:00:00 2001 From: yuenmichelle1 Date: Tue, 19 Dec 2023 10:49:10 -0600 Subject: [PATCH 2/4] add docs to aws for kinesis stream --- app/models/kinesis_stream.rb | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/app/models/kinesis_stream.rb b/app/models/kinesis_stream.rb index ad2b228..20ff422 100644 --- a/app/models/kinesis_stream.rb +++ b/app/models/kinesis_stream.rb @@ -21,7 +21,8 @@ def create_events(payload) ## ALSO NOTING: THIS CATCH, LOG, DEDUPE AND TRY UPSERTING AGAIN TO DB SITUATION IS TEMPORARY AND ONLY USED ## TO SEE WHAT THE DUPES IN THE KINESIS PAYLOAD ARE. ## ONE MORE NOTE: per Kinesis docs, it is VERY possible for Kinesis stream to send duplicates and - ## the recommendation of AWS is to appropriately handle process records. + ## the recommendation of AWS is to appropriately handle process records. + ## SEE: https://docs.aws.amazon.com/streams/latest/dev/kinesis-record-processor-duplicates.html upsert_comments unless @comment_events.empty? upsert_classifications unless @classification_events.empty? From 34dac8df932b35537f53ee2cd5528df084d2c880 Mon Sep 17 00:00:00 2001 From: yuenmichelle1 Date: Tue, 19 Dec 2023 13:22:32 -0600 Subject: [PATCH 3/4] update sentry to capture message --- app/models/kinesis_stream.rb | 3 +++ 1 file changed, 3 insertions(+) diff --git a/app/models/kinesis_stream.rb b/app/models/kinesis_stream.rb index 20ff422..fabadc4 100644 --- a/app/models/kinesis_stream.rb +++ b/app/models/kinesis_stream.rb @@ -42,6 +42,7 @@ def upsert_comments level: 'warning' ) Sentry.add_breadcrumb(crumb) + Sentry.capture_exception(e) if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') @comment_events = @comment_events.uniq { |comment| comment[:comment_id] } retry @@ -61,6 +62,7 @@ def upsert_classifications level: 'warning' ) Sentry.add_breadcrumb(crumb) + Sentry.capture_exception(e) if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') @classification_events = @classification_events.uniq { |classification| classification[:classification_id] } retry @@ -80,6 +82,7 @@ def upsert_classification_user_groups level: 'warning' ) Sentry.add_breadcrumb(crumb) + Sentry.capture_exception(e) if e.message.include?('ON CONFLICT DO UPDATE command cannot affect row a second time') @classification_user_groups = @classification_user_groups.uniq { |cug| [cug[:classification_id], cug[:user_group_id]] } retry From 3f48078cf0433c9ed3049ebae3b20ae7d9bf6693 Mon Sep 17 00:00:00 2001 From: yuenmichelle1 Date: Tue, 19 Dec 2023 13:38:53 -0600 Subject: [PATCH 4/4] add spec to check if it at least dedupes --- spec/models/kinesis_stream_spec.rb | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/spec/models/kinesis_stream_spec.rb b/spec/models/kinesis_stream_spec.rb index 56afbf9..2be5a1d 100644 --- a/spec/models/kinesis_stream_spec.rb +++ b/spec/models/kinesis_stream_spec.rb @@ -71,5 +71,17 @@ end.to change(ClassificationUserGroup, :count).from(0).to(1) end end + + context 'payload has duplicates' do + it 'dedupes classification_events' do + kinesis_stream.create_events([classification_payload, classification_payload]) + expect(kinesis_stream.instance_variable_get(:@classification_events).length).to eq(1) + end + + it 'dedupes comment_events' do + kinesis_stream.create_events([comment_payload, comment_payload, comment_payload]) + expect(kinesis_stream.instance_variable_get(:@comment_events).length).to eq(1) + end + end end end