jonatas/poc_downloads.rb

## poc_downloads.rb
require 'bundler/inline'

gemfile do
  source 'https://rubygems.org'
  gem 'timescaledb'
  gem 'bulk_insert'
  gem 'pry'
end

require 'timescaledb'

class Download < ActiveRecord::Base
  acts_as_hypertable time_column: 'ts'

  scope :time_bucket, -> (range='1m', query="count(*)") do
    select("time_bucket('#{range}', #{time_column}) as #{time_column}, #{query}")
  end

  scope :per_minute, -> (query="count(*) as downloads") do
    time_bucket('1m', query).group(1)
  end

  scope :stats_agg_per_minute, -> do
    per_minute("gem_name, gem_version, stats_agg(cast(1.0 as double precision)) as stats_agg")
      .group(1,2,3)
  end

  scope :gems_per_minute, -> do
    per_minute("gem_name, count(*) as downloads").group(1,2)
  end

  scope :versions_per_minute, -> do
    per_minute("gem_name, gem_version, count(*) as downloads").group(1,2,3)
  end

  cagg = -> (view_name) do
    Class.new(ActiveRecord::Base) do
      self.table_name = "downloads_#{view_name}"

      QUERIES = {
        sum_downloads: "sum(downloads)::bigint as downloads",
        avg_downloads: "avg(downloads)::bigint as avg_downloads",
        rollup: "rollup(stats_agg) as stats_agg",
        rolling_downloads: "rolling(stats_agg)->sum() as downloads",
        rollup_downloads: "rollup(stats_agg)->sum() as downloads",
        rollup_downloads_per_gem: "gem_name,rollup(stats_agg)->sum() as downloads",
        rollup_downloads_per_version: "gem_name, gem_version, rollup(stats_agg)->sum() as downloads"
      }

      scope :rollup, -> (range='1d', query=:sum_downloads) do
        select("time_bucket('#{range}', ts) as ts, #{QUERIES[query] || query}")
          .group(1)
      end


      scope :per_hour, -> (query=:sum_downloads) do
        rollup('1h', query)
      end

      scope :per_day, -> (query=:sum_downloads) do
        rollup('1d', query)
      end

      scope :per_week, -> (query=:sum_downloads) do
        rollup('1w', query)
      end

      scope :per_month, -> (query=:sum_downloads) do
        rollup('1mon', query)
      end

      scope :per_year, -> (query=:sum_downloads) do
        rollup('1y', query)
      end

      def readonly?
        true
      end

      def self.refresh!
        ActiveRecord::Base.connection.execute <<-SQL
        CALL refresh_continuous_aggregate('#{table_name}', null, null);
        SQL
      end
    end
  end

  PerMinute = cagg['per_minute']
  PerHour= cagg['per_hour']
  PerDay = cagg['per_day']
  PerMonth = cagg['per_month']
  GemsPerMinute = cagg['gems_per_minute']
  GemsPerHour= cagg['gems_per_hour']
  GemsPerDay = cagg['gems_per_day']
  GemsPerMonth= cagg['gems_per_month']
  VersionsPerMinute= cagg['versions_per_minute']
  VersionsPerHour = cagg['versions_per_hour']
  VersionsPerDay = cagg['versions_per_day']
  VersionsPerMonth = cagg['versions_per_month']

  StatsAggPerMinute = cagg['stats_agg_per_minute']
  StatsAggPerHour = cagg['stats_agg_per_hour']
  StatsAggPerDay = cagg['stats_agg_per_day']
  StatsAggPerMonth = cagg['stats_agg_per_month']
end

# Connect to the database

ActiveRecord::Base.establish_connection(ENV['DATABASE_URL'])
ActiveRecord::Base.connection.instance_exec do
  ActiveRecord::Base.logger = Logger.new(STDOUT)

  %w[day hour minute].each do |frame|
    ["downloads_per_#{frame}",
     "downloads_gems_per_#{frame}",
     "downloads_versions_per_#{frame}",
     "downloads_stats_agg_per_#{frame}"
    ].each do |view|
      execute("DROP MATERIALIZED VIEW IF EXISTS #{view} cascade")
    end
  end

  drop_table(:downloads, force: :cascade) if Download.table_exists?

  hypertable_options = {
    time_column: 'ts',
    chunk_time_interval: '1 day',
    compress_segmentby: 'gem_name, gem_version',
    compress_orderby: 'ts DESC',
    compression_interval: '7 days'
  }

  create_table(:downloads, id: false, hypertable: hypertable_options) do |t|
    t.timestamptz :ts, null: false
    t.text :gem_name, :gem_version, null: false
    t.jsonb :payload
  end

  {
    per_minute: Download.per_minute,
    per_hour: Download::PerMinute.per_hour(:sum_downloads).group(1),
    per_day: Download::PerHour.per_day(:sum_downloads).group(1),
    per_month: Download::PerDay.per_month(:sum_downloads).group(1),

    gems_per_minute: Download.gems_per_minute,
    gems_per_hour: Download::GemsPerMinute.per_hour("gem_name, count(*) as downloads").group(1,2),
    gems_per_day: Download::GemsPerHour.per_day("gem_name, count(*) as downloads").group(1,2),
    gems_per_month: Download::GemsPerDay.per_month("gem_name, count(*) as downloads").group(1,2),

    versions_per_minute: Download.versions_per_minute,
    versions_per_hour: Download::VersionsPerMinute.per_hour("gem_name, gem_version, count(*) as downloads").group(1,2,3),
    versions_per_day: Download::VersionsPerHour.per_day("gem_name, gem_version, count(*) as downloads").group(1,2,3),
    versions_per_month: Download::VersionsPerDay.per_month("gem_name, gem_version, count(*) as downloads").group(1,2,3),

    stats_agg_per_minute: Download.stats_agg_per_minute,
    stats_agg_per_hour: Download::StatsAggPerMinute.per_hour(:rollup).group(1),
    stats_agg_per_day: Download::StatsAggPerHour.per_day(:rollup).group(1),
    stats_agg_per_month: Download::StatsAggPerDay.per_month(:rollup).group(1)
  }.each do |name, scope|
    puts "Creating continuous aggregate #{name}", scope.to_sql
    frame = name.to_s.split('per_').last
    create_continuous_aggregate(
      "downloads_#{name}",
      scope.to_sql,
      refresh_policies: {
        schedule_interval: "INTERVAL '1 #{frame}'",
        start_offset: "INTERVAL '3 #{frame}'",
        end_offset: "INTERVAL '1 minute'"
      })
  end
end

ActiveRecord::Base.logger = nil

PATH_PATTERN = /\/gems\/(?<gem_name>.*)-(?<gem_version>\d+.*)\.gem/

def parse_file(file)
  downloads = []
  File.open(file).each_line do |log_line|
    fragments = log_line.split
    path, response_code = fragments[10, 2]
    case response_code.to_i
      # Only count successful downloads
      # NB: we consider a 304 response a download attempt
    when 200, 304
      m = path.match(PATH_PATTERN)
      gem_name = m[:gem_name] || path
      gem_version = m[:gem_version]
      ip = fragments[3]
      ts = Time.parse fragments[4..9].join(' ')
      env = parse_env fragments[12..-1]

      payload = {ip:, env:}

      downloads << {ts:, gem_name:, gem_version:, payload:}
      if downloads.size == 5000
        insert_downloads(downloads)
        downloads.clear
      end
    end
  end

  if downloads.any?
    insert_downloads(downloads)
  end
end


# example env = "bundler/2.5.9 rubygems/3.3.25 ruby/3.1.0"
# output = {bundler: "2.5.9", rubygems: "3.3.25", ruby: "3.1.0"}
# case it says single word like jruby it appends true as the value
# example env = "jruby"
# output = {jruby: "true"}
def parse_env(output)
  env = output.join(' ').gsub(/command.*|\(.*\)|Ruby, /,'').strip
  env = nil if env == "(null)"
  env = env.split(' ').map do |info|
    pair = info.split(/\/|-/,2)
    pair << "true" if pair.size == 1
    pair
  end.to_h
end

def insert_downloads(downloads)
  Download.bulk_insert values: downloads
end

s3_files = [
  "2024-04-26T00_15_00.000-szTpTn9sP1-116Ajwl4N.log"
]

Benchmark.bm do |x|
  s3_files.each do |file|
    x.report "parse and load #{file}" do
      parse_file(file)
    end

    %w[
      PerMinute GemsPerMinute VersionsPerMinute StatsAggPerMinute
      PerHour   GemsPerHour   VersionsPerHour   StatsAggPerHour
      PerDay    GemsPerDay    VersionsPerDay    StatsAggPerDay
      PerMonth  GemsPerMonth  VersionsPerMonth  StatsAggPerMonth
    ].each do |view|
        x.report "Refresh #{view}" do
          Download.const_get(view).refresh!
        end
      end
    end
end

require "pry";binding.pry
=begin
Download::PerHour.first
Download::GemsPerHour.all
Download::VersionsPerHour.where(gem_name: "rails").pluck(:gem_version, :downloads) # => [["1.2.3.4", 6], ["6.1.7", 1], ["7.0.2", 1]]
=end
	require 'bundler/inline'

	gemfile do
	source 'https://rubygems.org'
	gem 'timescaledb'
	gem 'bulk_insert'
	gem 'pry'
	end

	require 'timescaledb'

	class Download < ActiveRecord::Base
	acts_as_hypertable time_column: 'ts'

	scope :time_bucket, -> (range='1m', query="count(*)") do
	select("time_bucket('#{range}', #{time_column}) as #{time_column}, #{query}")
	end

	scope :per_minute, -> (query="count(*) as downloads") do
	time_bucket('1m', query).group(1)
	end

	scope :stats_agg_per_minute, -> do
	per_minute("gem_name, gem_version, stats_agg(cast(1.0 as double precision)) as stats_agg")
	.group(1,2,3)
	end

	scope :gems_per_minute, -> do
	per_minute("gem_name, count(*) as downloads").group(1,2)
	end

	scope :versions_per_minute, -> do
	per_minute("gem_name, gem_version, count(*) as downloads").group(1,2,3)
	end

	cagg = -> (view_name) do
	Class.new(ActiveRecord::Base) do
	self.table_name = "downloads_#{view_name}"

	QUERIES = {
	sum_downloads: "sum(downloads)::bigint as downloads",
	avg_downloads: "avg(downloads)::bigint as avg_downloads",
	rollup: "rollup(stats_agg) as stats_agg",
	rolling_downloads: "rolling(stats_agg)->sum() as downloads",
	rollup_downloads: "rollup(stats_agg)->sum() as downloads",
	rollup_downloads_per_gem: "gem_name,rollup(stats_agg)->sum() as downloads",
	rollup_downloads_per_version: "gem_name, gem_version, rollup(stats_agg)->sum() as downloads"
	}

	scope :rollup, -> (range='1d', query=:sum_downloads) do
	select("time_bucket('#{range}', ts) as ts, #{QUERIES[query] \|\| query}")
	.group(1)
	end


	scope :per_hour, -> (query=:sum_downloads) do
	rollup('1h', query)
	end

	scope :per_day, -> (query=:sum_downloads) do
	rollup('1d', query)
	end

	scope :per_week, -> (query=:sum_downloads) do
	rollup('1w', query)
	end

	scope :per_month, -> (query=:sum_downloads) do
	rollup('1mon', query)
	end

	scope :per_year, -> (query=:sum_downloads) do
	rollup('1y', query)
	end

	def readonly?
	true
	end

	def self.refresh!
	ActiveRecord::Base.connection.execute <<-SQL
	CALL refresh_continuous_aggregate('#{table_name}', null, null);
	SQL
	end
	end
	end

	PerMinute = cagg['per_minute']
	PerHour= cagg['per_hour']
	PerDay = cagg['per_day']
	PerMonth = cagg['per_month']
	GemsPerMinute = cagg['gems_per_minute']
	GemsPerHour= cagg['gems_per_hour']
	GemsPerDay = cagg['gems_per_day']
	GemsPerMonth= cagg['gems_per_month']
	VersionsPerMinute= cagg['versions_per_minute']
	VersionsPerHour = cagg['versions_per_hour']
	VersionsPerDay = cagg['versions_per_day']
	VersionsPerMonth = cagg['versions_per_month']

	StatsAggPerMinute = cagg['stats_agg_per_minute']
	StatsAggPerHour = cagg['stats_agg_per_hour']
	StatsAggPerDay = cagg['stats_agg_per_day']
	StatsAggPerMonth = cagg['stats_agg_per_month']
	end

	# Connect to the database

	ActiveRecord::Base.establish_connection(ENV['DATABASE_URL'])
	ActiveRecord::Base.connection.instance_exec do
	ActiveRecord::Base.logger = Logger.new(STDOUT)

	%w[day hour minute].each do \|frame\|
	["downloads_per_#{frame}",
	"downloads_gems_per_#{frame}",
	"downloads_versions_per_#{frame}",
	"downloads_stats_agg_per_#{frame}"
	].each do \|view\|
	execute("DROP MATERIALIZED VIEW IF EXISTS #{view} cascade")
	end
	end

	drop_table(:downloads, force: :cascade) if Download.table_exists?

	hypertable_options = {
	time_column: 'ts',
	chunk_time_interval: '1 day',
	compress_segmentby: 'gem_name, gem_version',
	compress_orderby: 'ts DESC',
	compression_interval: '7 days'
	}

	create_table(:downloads, id: false, hypertable: hypertable_options) do \|t\|
	t.timestamptz :ts, null: false
	t.text :gem_name, :gem_version, null: false
	t.jsonb :payload
	end

	{
	per_minute: Download.per_minute,
	per_hour: Download::PerMinute.per_hour(:sum_downloads).group(1),
	per_day: Download::PerHour.per_day(:sum_downloads).group(1),
	per_month: Download::PerDay.per_month(:sum_downloads).group(1),

	gems_per_minute: Download.gems_per_minute,
	gems_per_hour: Download::GemsPerMinute.per_hour("gem_name, count(*) as downloads").group(1,2),
	gems_per_day: Download::GemsPerHour.per_day("gem_name, count(*) as downloads").group(1,2),
	gems_per_month: Download::GemsPerDay.per_month("gem_name, count(*) as downloads").group(1,2),

	versions_per_minute: Download.versions_per_minute,
	versions_per_hour: Download::VersionsPerMinute.per_hour("gem_name, gem_version, count(*) as downloads").group(1,2,3),
	versions_per_day: Download::VersionsPerHour.per_day("gem_name, gem_version, count(*) as downloads").group(1,2,3),
	versions_per_month: Download::VersionsPerDay.per_month("gem_name, gem_version, count(*) as downloads").group(1,2,3),

	stats_agg_per_minute: Download.stats_agg_per_minute,
	stats_agg_per_hour: Download::StatsAggPerMinute.per_hour(:rollup).group(1),
	stats_agg_per_day: Download::StatsAggPerHour.per_day(:rollup).group(1),
	stats_agg_per_month: Download::StatsAggPerDay.per_month(:rollup).group(1)
	}.each do \|name, scope\|
	puts "Creating continuous aggregate #{name}", scope.to_sql
	frame = name.to_s.split('per_').last
	create_continuous_aggregate(
	"downloads_#{name}",
	scope.to_sql,
	refresh_policies: {
	schedule_interval: "INTERVAL '1 #{frame}'",
	start_offset: "INTERVAL '3 #{frame}'",
	end_offset: "INTERVAL '1 minute'"
	})
	end
	end

	ActiveRecord::Base.logger = nil

	PATH_PATTERN = /\/gems\/(?<gem_name>.)-(?<gem_version>\d+.)\.gem/

	def parse_file(file)
	downloads = []
	File.open(file).each_line do \|log_line\|
	fragments = log_line.split
	path, response_code = fragments[10, 2]
	case response_code.to_i
	# Only count successful downloads
	# NB: we consider a 304 response a download attempt
	when 200, 304
	m = path.match(PATH_PATTERN)
	gem_name = m[:gem_name] \|\| path
	gem_version = m[:gem_version]
	ip = fragments[3]
	ts = Time.parse fragments[4..9].join(' ')
	env = parse_env fragments[12..-1]

	payload = {ip:, env:}

	downloads << {ts:, gem_name:, gem_version:, payload:}
	if downloads.size == 5000
	insert_downloads(downloads)
	downloads.clear
	end
	end
	end

	if downloads.any?
	insert_downloads(downloads)
	end
	end


	# example env = "bundler/2.5.9 rubygems/3.3.25 ruby/3.1.0"
	# output = {bundler: "2.5.9", rubygems: "3.3.25", ruby: "3.1.0"}
	# case it says single word like jruby it appends true as the value
	# example env = "jruby"
	# output = {jruby: "true"}
	def parse_env(output)
	env = output.join(' ').gsub(/command.\|\(.\)\|Ruby, /,'').strip
	env = nil if env == "(null)"
	env = env.split(' ').map do \|info\|
	pair = info.split(/\/\|-/,2)
	pair << "true" if pair.size == 1
	pair
	end.to_h
	end

	def insert_downloads(downloads)
	Download.bulk_insert values: downloads
	end

	s3_files = [
	"2024-04-26T00_15_00.000-szTpTn9sP1-116Ajwl4N.log"
	]

	Benchmark.bm do \|x\|
	s3_files.each do \|file\|
	x.report "parse and load #{file}" do
	parse_file(file)
	end

	%w[
	PerMinute GemsPerMinute VersionsPerMinute StatsAggPerMinute
	PerHour GemsPerHour VersionsPerHour StatsAggPerHour
	PerDay GemsPerDay VersionsPerDay StatsAggPerDay
	PerMonth GemsPerMonth VersionsPerMonth StatsAggPerMonth
	].each do \|view\|
	x.report "Refresh #{view}" do
	Download.const_get(view).refresh!
	end
	end
	end
	end

	require "pry";binding.pry
	=begin
	Download::PerHour.first
	Download::GemsPerHour.all
	Download::VersionsPerHour.where(gem_name: "rails").pluck(:gem_version, :downloads) # => [["1.2.3.4", 6], ["6.1.7", 1], ["7.0.2", 1]]
	=end