Implement the rest of the VSD count metrics

This change adds the group, pipeline and MR count metrics to the Value
Stream Dashboard count service.

app/models/concerns/each_batch.rb

@@ -219,6 +219,7 @@ def each_batch_count(of: 1000, column: :id, last_count: 0, last_value: nil)
           new_count, last_value =
             unscoped
             .from(inner_query)
+            .unscope(where: :type)
             .order(count: :desc)
             .limit(1)
             .pick(:count, column)

ee/app/models/analytics/value_stream_dashboard/count.rb

@@ -14,7 +14,7 @@ class Count < ApplicationRecord
 
       validates :namespace_id, :count, :metric, :recorded_at, presence: true
 
-      enum metric: { projects: 1, issues: 2 }
+      enum metric: { projects: 1, issues: 2, groups: 3, merge_requests: 4, pipelines: 5 }
     end
   end
 end

ee/app/services/analytics/value_stream_dashboard/top_level_group_counter_service.rb

@@ -7,20 +7,26 @@ class TopLevelGroupCounterService
       INSERT_BATCH_SIZE = 100
       COUNT_BATCH_SIZE = 5000
 
+      # rubocop: disable CodeReuse/ActiveRecord
+      GROUP_SELECT_SCOPE = ->(scope) { scope.select(:id) }
+      PROJECT_SELECT_SCOPE = ->(scope) {
+        scope.joins(:project).select(:id, Project.arel_table[:id].as('tmp_project_id'))
+      }
+
       # - metric: metric value for the metrics enum in Analytics::ValueStreamDashboard::Count
-      # - namespace_scope: scope for iterating (EachBatch) over the namespaces (Group or ProjectNamespace)
-      # - inner_namespace_query: transform the yielded query provided by the namespace_scope
+      # - namespace_class: class to use for iterating (EachBatch) over specific namespaces (Group or ProjectNamespace)
+      # - inner_namespace_query: transform the yielded query provided by the namespace_class
       #
       #  Example:
       #
       #  Consider the following configuration:
       #  ```
-      #    namespace_scope = ->{ Group.where('traversal_ids[1] = ?', group_id) }
+      #    namespace_class = Group
       #    inner_namespace_query = ->(scope) { scope.select(:id) }
       #
       #    # the two configuration option will be used in EachBatch:
       #
-      #    namespace_scope.each_batch do |relation|
+      #    namespace_class.where("traversal_ids[1] = ?", 1).each_batch do |relation|
       #      # the `inner_namespace_query` lambda will modify the yielded relation, result:
       #      namespaces = relation.select(:id)
       #    end
@@ -28,23 +34,41 @@ class TopLevelGroupCounterService
       #
       # - count_scope: use this scope when counting items in batches
       # - count_batching_column: batch countable items by this column
-      # rubocop: disable CodeReuse/ActiveRecord
       COUNTS_TO_COLLECT = {
         projects: {
           metric: ::Analytics::ValueStreamDashboard::Count.metrics[:projects],
-          namespace_scope: ->(group_id) { Group.where('traversal_ids[1] = ?', group_id) },
-          inner_namespace_query: ->(scope) { scope.select(:id) },
+          namespace_class: Group,
+          inner_namespace_query: GROUP_SELECT_SCOPE,
           count_scope: Project.method(:in_namespace),
           count_batching_column: :id
         }.freeze,
         issues: {
           metric: ::Analytics::ValueStreamDashboard::Count.metrics[:issues],
-          namespace_scope: ->(group_id) { Namespaces::ProjectNamespace.where('traversal_ids[1] = ?', group_id) },
-          inner_namespace_query: ->(scope) {
-            scope.joins(:project).select(:id, Project.arel_table[:id].as('tmp_project_id'))
-          },
+          namespace_class: Namespaces::ProjectNamespace,
+          inner_namespace_query: PROJECT_SELECT_SCOPE,
           count_scope: ->(namespace) { Issue.in_projects(namespace.tmp_project_id) },
           count_batching_column: :iid
+        }.freeze,
+        groups: {
+          metric: ::Analytics::ValueStreamDashboard::Count.metrics[:groups],
+          namespace_class: Group,
+          inner_namespace_query: GROUP_SELECT_SCOPE,
+          count_batching_column: :id,
+          count_scope: ->(namespace) { Group.where(parent_id: namespace.id) }
+        }.freeze,
+        merge_requests: {
+          metric: ::Analytics::ValueStreamDashboard::Count.metrics[:merge_requests],
+          namespace_class: Namespaces::ProjectNamespace,
+          inner_namespace_query: PROJECT_SELECT_SCOPE,
+          count_batching_column: :iid,
+          count_scope: ->(namespace) { MergeRequest.where(target_project_id: namespace.tmp_project_id) }
+        }.freeze,
+        pipelines: {
+          metric: ::Analytics::ValueStreamDashboard::Count.metrics[:pipelines],
+          namespace_class: Namespaces::ProjectNamespace,
+          inner_namespace_query: PROJECT_SELECT_SCOPE,
+          count_batching_column: :id,
+          count_scope: ->(namespace) { Ci::Pipeline.where(project_id: namespace.tmp_project_id) }
         }.freeze
       }.freeze
       # rubocop: enable CodeReuse/ActiveRecord
@@ -89,7 +113,7 @@ def self.load_cursor(raw_cursor:, countable_config: nil)
         countable_config ||= COUNTS_TO_COLLECT.values.detect { |config| config[:metric] == raw_cursor[:metric] }
 
         Gitlab::Analytics::ValueStreamDashboard::NamespaceCursor.new(
-          namespace_scope: countable_config[:namespace_scope],
+          namespace_class: countable_config[:namespace_class],
           inner_namespace_query: countable_config[:inner_namespace_query],
           cursor_data: raw_cursor.merge({ metric: countable_config[:metric] })
         )

ee/lib/gitlab/analytics/value_stream_dashboard/namespace_cursor.rb

@@ -6,8 +6,8 @@ module ValueStreamDashboard
       class NamespaceCursor
         NAMESPACE_BATCH_SIZE = 300
 
-        def initialize(namespace_scope:, inner_namespace_query:, cursor_data:)
-          @namespace_scope = namespace_scope
+        def initialize(namespace_class:, inner_namespace_query:, cursor_data:)
+          @namespace_class = namespace_class
           @inner_namespace_query = inner_namespace_query
           @cursor_data = cursor_data
           @top_level_namespace_id = cursor_data.fetch(:top_level_namespace_id)
@@ -35,12 +35,12 @@ def dump
 
         private
 
-        attr_reader :namespace_scope, :inner_namespace_query, :cursor_data, :top_level_namespace_id
+        attr_reader :namespace_class, :inner_namespace_query, :cursor_data, :top_level_namespace_id
 
         # rubocop: disable CodeReuse/ActiveRecord
         def enumerator
           @enumerator ||= begin
-            scope = namespace_scope.call(top_level_namespace_id)
+            scope = namespace_class.where('traversal_ids[1] = ?', top_level_namespace_id)
             if cursor_data[:namespace_id]
               scope = scope.where(Namespace.arel_table[:id].gteq(cursor_data[:namespace_id]))
             end

ee/spec/lib/gitlab/analytics/value_stream_dashboard/namespace_cursor_spec.rb

@@ -12,7 +12,7 @@
 
   let(:cursor) do
     described_class.new(
-      namespace_scope: ->(group_id) { Group.where('traversal_ids[1] = ?', group_id) },
+      namespace_class: Group,
       inner_namespace_query: ->(namespaces) { namespaces.select('id AS custom_column') },
       cursor_data: cursor_data
     )

ee/spec/services/analytics/value_stream_dashboard/top_level_group_counter_service_spec.rb

@@ -16,6 +16,13 @@
   let_it_be(:another_issue_in_group) { create(:issue, project: project_in_group) }
   let_it_be(:issue_in_subsubgroup) { create(:issue, project: project_in_subsubgroup) }
 
+  let_it_be(:mr_in_subsubgroup1) { create(:merge_request, :unique_branches, source_project: project_in_subsubgroup) }
+  let_it_be(:mr_in_subsubgroup2) { create(:merge_request, :unique_branches, source_project: project_in_subsubgroup) }
+
+  let_it_be(:ci_pipeline1) { create(:ci_pipeline, project: project_in_group) }
+  let_it_be(:ci_pipeline2) { create(:ci_pipeline, project: project_in_group) }
+  let_it_be(:ci_pipeline3) { create(:ci_pipeline, project: project_in_group) }
+
   let_it_be(:aggregation) { create(:value_stream_dashboard_aggregation, namespace: group, last_run_at: nil) }
 
   let(:raw_cursor) { { top_level_namespace_id: aggregation.id } }
@@ -38,7 +45,18 @@
       { metric: 'issues', namespace_id: project_in_group.project_namespace.id, count: 2 },
       { metric: 'issues', namespace_id: project_in_subgroup.project_namespace.id, count: 0 },
       { metric: 'issues', namespace_id: another_project_in_subgroup.project_namespace.id, count: 0 },
-      { metric: 'issues', namespace_id: project_in_subsubgroup.project_namespace.id, count: 1 }
+      { metric: 'issues', namespace_id: project_in_subsubgroup.project_namespace.id, count: 1 },
+      { metric: 'groups', namespace_id: group.id, count: 1 },
+      { metric: 'groups', namespace_id: subgroup.id, count: 1 },
+      { metric: 'groups', namespace_id: subsubgroup.id, count: 0 },
+      { metric: 'merge_requests', namespace_id: project_in_group.project_namespace.id, count: 0 },
+      { metric: 'merge_requests', namespace_id: project_in_subgroup.project_namespace.id, count: 0 },
+      { metric: 'merge_requests', namespace_id: another_project_in_subgroup.project_namespace.id, count: 0 },
+      { metric: 'merge_requests', namespace_id: project_in_subsubgroup.project_namespace.id, count: 2 },
+      { metric: 'pipelines', namespace_id: project_in_group.project_namespace.id, count: 3 },
+      { metric: 'pipelines', namespace_id: project_in_subgroup.project_namespace.id, count: 0 },
+      { metric: 'pipelines', namespace_id: another_project_in_subgroup.project_namespace.id, count: 0 },
+      { metric: 'pipelines', namespace_id: project_in_subsubgroup.project_namespace.id, count: 0 }
     ]
   end
 
@@ -97,7 +115,8 @@
     it 'continues the processing from the cursor' do
       run_service
 
-      expect(persisted_counts).to match(expected_counts.map { |a| have_attributes(a) })
+      issues_only = persisted_counts.select(&:issues?)
+      expect(issues_only).to match(expected_counts.map { |a| have_attributes(a) })
     end
   end