add download sparks

Author: ezekg

app/controllers/priv/analytics/gauges/downloads_controller.rb

@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Priv::Analytics::Gauges
+  class DownloadsController < Priv::Analytics::BaseController
+    use_clickhouse
+
+    CACHE_TTL      = 10.minutes
+    CACHE_RACE_TTL = 1.minute
+
+    typed_query {
+      param :product, type: :uuid, optional: true, as: :product_id
+      param :package, type: :uuid, optional: true, as: :package_id
+      param :release, type: :uuid, optional: true, as: :release_id
+    }
+    def show
+      authorize! with: Accounts::AnalyticsPolicy
+
+      gauge = Analytics::Gauge.new(
+        :downloads,
+        **download_query,
+      )
+
+      unless gauge.valid?
+        render_bad_request *gauge.errors.as_jsonapi(
+          title: 'Bad request',
+          source: :parameter,
+          sources: {
+            parameters: {
+              product_id: 'product',
+              package_id: 'package',
+              release_id: 'release',
+            },
+          },
+        )
+
+        return
+      end
+
+      data = Rails.cache.fetch gauge.cache_key, expires_in: CACHE_TTL, race_condition_ttl: CACHE_RACE_TTL do
+        gauge.as_json
+      end
+
+      render json: { data: }
+    end
+  end
+end

app/controllers/priv/analytics/sparks/downloads_controller.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module Priv::Analytics::Sparks
+  class DownloadsController < Priv::Analytics::BaseController
+    use_clickhouse
+
+    CACHE_TTL      = 10.minutes
+    CACHE_RACE_TTL = 1.minute
+
+    typed_query {
+      param :product, type: :uuid, optional: true, as: :product_id
+      param :package, type: :uuid, optional: true, as: :package_id
+      param :release, type: :uuid, optional: true, as: :release_id
+      param :date, type: :hash, optional: true, collapse: { format: :child_parent } do
+        param :start, type: :date, coerce: true
+        param :end, type: :date, coerce: true
+      end
+    }
+    def show
+      authorize! with: Accounts::AnalyticsPolicy
+
+      series = Analytics::Series.new(
+        :downloads,
+        **download_query,
+      )
+
+      unless series.valid?
+        render_bad_request *series.errors.as_jsonapi(
+          title: 'Bad request',
+          source: :parameter,
+          sources: {
+            parameters: {
+              product_id: 'product',
+              package_id: 'package',
+              release_id: 'release',
+              start_date: 'date[start]',
+              end_date: 'date[end]',
+            },
+          },
+        )
+
+        return
+      end
+
+      data = Rails.cache.fetch series.cache_key, expires_in: CACHE_TTL, race_condition_ttl: CACHE_RACE_TTL do
+        series.as_json
+      end
+
+      render json: { data: }
+    end
+  end
+end

app/models/analytics/gauge.rb

@@ -11,6 +11,7 @@ class Gauge
 
     COUNTERS = {
       alus: ActiveLicensedUsers,
+      downloads: Downloads,
       licenses: Licenses,
       machines: Machines,
       users: Users,

app/models/analytics/gauge/downloads.rb

@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+module Analytics
+  class Gauge
+    class Downloads
+      def initialize(account:, environment:, product_id: nil, package_id: nil, release_id: nil)
+        @account     = account
+        @environment = environment
+        @product_id  = product_id
+        @package_id  = package_id
+        @release_id  = release_id
+      end
+
+      def metrics = %w[downloads]
+      def count
+        event_type_ids = EventType.by_pattern('artifact.downloaded')
+                                  .collect(&:id)
+        return {} if
+          event_type_ids.empty?
+
+        scope = EventLog::Clickhouse.for_account(account)
+                                    .for_environment(environment)
+                                    .where(
+                                      event_type_id: event_type_ids,
+                                      created_date: Date.current,
+                                    )
+                                    .where(
+                                      'metadata.product IS NOT NULL',
+                                    )
+
+        unless product_id.nil?
+          scope = scope.where(
+            Arel.sql('metadata.product.:String') => product_id,
+          )
+        end
+
+        unless package_id.nil?
+          scope = scope.where(
+            Arel.sql('metadata.package.:String') => package_id,
+          )
+        end
+
+        unless release_id.nil?
+          scope = scope.where(
+            Arel.sql('metadata.release.:String') => release_id,
+          )
+        end
+
+        count = scope.count
+
+        { 'downloads' => count }.reject { _2.zero? }
+      end
+
+      private
+
+      attr_reader :account, :environment, :product_id, :package_id, :release_id
+    end
+  end
+end

app/models/analytics/series.rb

@@ -7,6 +7,7 @@ class Series
     Bucket = Data.define(:metric, :date, :count)
 
     COUNTERS = {
+      downloads: Sparks::Downloads,
       events: Events,
       requests: Requests,
       sparks: Sparks,

app/models/analytics/series/sparks/downloads.rb

@@ -0,0 +1,71 @@
+# frozen_string_literal: true
+
+module Analytics
+  class Series
+    class Sparks
+      class Downloads
+        def initialize(account:, environment:, product_id: nil, package_id: nil, release_id: nil, realtime: true, **)
+          @account     = account
+          @environment = environment
+          @product_id  = product_id
+          @package_id  = package_id
+          @release_id  = release_id
+          @realtime    = realtime
+        end
+
+        def metrics = %w[downloads]
+        def count(start_date:, end_date:)
+          scope = ReleaseDownloadSpark.for_account(account)
+                                      .for_environment(environment)
+                                      .where(
+                                        created_date: start_date..end_date,
+                                      )
+
+          unless product_id.nil?
+            scope = scope.where(product_id:)
+          end
+
+          unless package_id.nil?
+            scope = scope.where(package_id:)
+          end
+
+          unless release_id.nil?
+            scope = scope.where(release_id:)
+          end
+
+          rows = scope.group(:created_date)
+                      .pluck(
+                        :created_date,
+                        Arel.sql('sum(count)'),
+                      )
+
+          counts = rows.each_with_object({}) do |(date, count), hash|
+            hash[['downloads', date]] = count
+          end
+
+          # defer to gauge for a realtime count since sparks are nightly
+          if realtime? && end_date.today?
+            gauge = Analytics::Gauge.new(:downloads, account:, environment:, product_id:, package_id:, release_id:)
+
+            gauge.measurements.each do |measurement|
+              counts[[measurement.metric, end_date]] = measurement.count
+            end
+          end
+
+          counts
+        end
+
+        private
+
+        attr_reader :account,
+                    :environment,
+                    :product_id,
+                    :package_id,
+                    :release_id,
+                    :realtime
+
+        def realtime? = !!realtime
+      end
+    end
+  end
+end

app/models/release_download_spark.rb

@@ -0,0 +1,8 @@
+# frozen_string_literal: true
+
+class ReleaseDownloadSpark < ClickhouseRecord
+  include Accountable, Environmental
+
+  has_environment
+  has_account
+end

app/services/broadcast_event_service.rb

@@ -51,7 +51,7 @@ def call
               { product: resource.product_id, package: resource.package_id, prev: meta[:current], next: meta[:next] }
             when /^artifact\.downloaded$/,
                  /^release\.downloaded$/
-              { product: resource.product_id, package: resource.package_id, version: resource.version }
+              { product: resource.product_id, package: resource.package_id, release: resource.release_id, version: resource.version }
             when /^license\.validation\./
               { code: meta[:code] }
             when /\.updated$/

app/workers/record_release_download_spark_worker.rb

@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+class RecordReleaseDownloadSparkWorker < BaseWorker
+  sidekiq_options queue: :cron,
+                  cronitor_enabled: true
+
+  def perform(account_id)
+    event_type_ids = EventType.by_pattern('artifact.downloaded')
+                              .collect(&:id)
+    return if
+      event_type_ids.empty?
+
+    events_cte = EventLog::Clickhouse.where(account_id:, created_date: Date.yesterday, event_type_id: event_type_ids)
+                                     .where('metadata.product IS NOT NULL')
+                                     .select(
+                                       :account_id,
+                                       :environment_id,
+                                       :created_date,
+                                       :created_at,
+                                       'metadata.product.:String AS product_id',
+                                       'metadata.package.:String AS package_id',
+                                       'metadata.release.:String AS release_id',
+                                     )
+
+    agg_cte = EventLog::Clickhouse.from('release_download_events')
+                                  .select(
+                                    :account_id,
+                                    :environment_id,
+                                    :created_date,
+                                    'max(created_at) AS created_at',
+                                    :product_id,
+                                    :package_id,
+                                    :release_id,
+                                    'count() AS count',
+                                  )
+                                  .group(
+                                    :account_id,
+                                    :environment_id,
+                                    :created_date,
+                                    :product_id,
+                                    :package_id,
+                                    :release_id,
+                                  )
+
+    ReleaseDownloadSpark.connection.execute(<<~SQL.squish)
+      WITH
+        release_download_events AS (#{events_cte.to_sql}),
+        release_download_agg    AS (#{agg_cte.to_sql})
+      INSERT INTO release_download_sparks (
+        account_id,
+        environment_id,
+        created_date,
+        created_at,
+        product_id,
+        package_id,
+        release_id,
+        count
+      )
+      SELECT
+        account_id,
+        environment_id,
+        created_date,
+        created_at,
+        product_id,
+        package_id,
+        release_id,
+        count
+      FROM
+        release_download_agg
+    SQL
+  end
+end

app/workers/record_release_download_sparks_worker.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class RecordReleaseDownloadSparksWorker < BaseWorker
+  sidekiq_options queue: :cron,
+                  cronitor_enabled: true
+
+  def perform
+    Account.unordered.paid.find_each do |account|
+      jitter = rand(0..30.minutes) # prevent a thundering herd effect
+
+      RecordReleaseDownloadSparkWorker.perform_in(jitter, account.id)
+    end
+  end
+end