S3

gistfile1.txt

#!/usr/bin/env ruby
 require "aws-sdk"
 require "aws/s3"
  require 'parallel'
  require 'thor'
  require 'colorize'
  require 'ap'
  require 'syslog/logger'
  
  NUMBER =1
  THREADS=10
  IMAGE  ="logo.jpg"
  BUCKET ="prodiecdn"
  class Lambda < Thor
    

    desc "s3","s3 access repair"
    option :dryrun, :type => :boolean
    option :verbose, :type => :boolean, :aliases => "-v"
    option :threads, :type => :numeric, :default => THREADS
    option :bucket, :type => :string, :default => BUCKET
    option :access_key_id, :type => :string, :required => true
    option :secret_access_key, :type => :string, :required => true
    option :region, :type => :string, :default => 'eu-west-1'
    option :log, :type => :boolean, :default => false, :desc => 'to see logs journalctl  SYSLOG_IDENTIFIER=LAMBDA'
    def s3()
      begin
        # Initialize s3 object
        s3 = Aws::S3::Resource.new(
            :access_key_id => options[:access_key_id],
            :secret_access_key => options[:secret_access_key],
            :region => options[:region]
        ).bucket(options[:bucket])
        puts s3.object('bucketexplorerbucketdefaults_cpm-bast.xml').key
        puts "x"
        s3.object('bucketexplorerbucketdefaults_cpm-bast.xml').acl.put({ acl: :public_read})
        
      rescue Exception => e
        puts "Error"
        puts e.message
        exit 255
      end     
      fin=false
      semaphore1 = Mutex.new
      semaphore2 = Mutex.new
      puts "Listing S3 bucket "+options[:bucket].yellow+" with "+options[:threads].to_s.yellow+" threads"
      # iterate over listing
      list=[]
      tmp=[]
      nb_thread=0
      Thread.new{
        i=0
        s3.objects.each do |object|
          #puts "i"
            tmp.push object
            i+=1
            if (i > 3)
              
              semaphore1.synchronize{
                list.push(*tmp)
              }
              semaphore2.synchronize{
                puts " #{list.count} #{nb_thread}/ #{options[:threads]} "
              }
              
              
              tmp=[]
              i=0
            end
        end
        fin=true
      }

      
      

      Thread.new {
        
        while true
          semaphore2.synchronize {
            
            semaphore1.synchronize {
            if (list.count > 0) && nb_thread < options[:threads] 
                Thread.new {
                  semaphore2.synchronize {
                    nb_thread+=1
                  }
                  semaphore1.synchronize {
                    obj=list.pop
                    if options[:verbose]
                      puts obj.key
                    end
                  }
                  #puts obj
                  policy = AWS::S3::S3Object.acl(obj.key, options[:bucket])
                  policy.grants = [ AWS::S3::ACL::Grant.grant(:public_read) ]
                  AWS::S3::S3Object.acl(obj.key, options[:bucket], policy)
                  
                  
                  semaphore2.synchronize {
                    nb_thread-=1
                  }

                }
            end
            }
          }
          sleep 0.001
        end
      }



      while !fin
        sleep 2
      end

      

      


    end 
 
    desc "test","s3 access repair"
    option :dryrun, :type => :boolean
    option :verbose, :type => :boolean, :aliases => "-v"
    option :threads, :type => :numeric, :default => THREADS
    option :number, :type => :numeric, :default => NUMBER
    option :image, :type => :string, :default => IMAGE
    option :bucket, :type => :string, :default => BUCKET
    option :access_key_id, :type => :string, :required => true
    option :secret_access_key, :type => :string, :required => true
    option :region, :type => :string, :default => 'eu-west-1'
    option :log, :type => :boolean, :default => false, :desc => 'to see logs journalctl  SYSLOG_IDENTIFIER=LAMBDA'   
    def test()
      begin
        # Initialize s3 object
        s3 = Aws::S3::Resource.new(
            :access_key_id => options[:access_key_id],
            :secret_access_key => options[:secret_access_key],
            :region => options[:region]
        )
        s3 = s3.bucket('prod-image-worker')
      rescue
        puts "Error"
        exit 255
      end

      puts "Populating S3 bucket "+options[:bucket].yellow+" with "+options[:number].to_s.yellow+" images and "+options[:threads].to_s.yellow+" threads"
      puts "Image "+options[:image].yellow

      option={
          :in_threads=> options[:threads]
      }
      if ! options[:verbose]
        option[:progress] = "Populating S3..."
      end
      if options[:log]
        log = Syslog::Logger.new 'LAMBDA'
      end
      Parallel.each(1..options[:number], option) { |index|
        # FORMAT : [etude_id]_[year]_[month]_[day]_[hour][minutes][seconds]_[bucket_id]_[filename_md5].[extension]
        etude_id=index
        year='%04d' % (2005+Random.rand(10))
        day='%02d' % (Random.rand(29)+1)
        month='%02d' % (Random.rand(11)+1)
        hour='%02d' % (Random.rand(23)+1)
        minutes='%02d' % (Random.rand(58)+1)
        seconds='%02d' % (Random.rand(58)+1)
        bucket_id=index
        filename="#{etude_id}_#{year}_#{month}_#{day}_#{hour}#{minutes}#{seconds}_#{bucket_id}_000000000000.jpg"

        begin
          obj=s3.object(filename)
          File.open(options[:image], 'rb') do |file|
            if ! options[:dryrun]
              obj.put(body: file)
            else
              sleep 1
            end
          end
          if options[:verbose]
            puts "\t"+filename+" Ok.".green+"\n"
          end
          if options[:log]
            log.info filename+" Ok."
          end
        rescue Exception => e
          ap e
          puts "\t"+filename+" ERROR !".red
          if options[:log]
            log.error filename+" ERROR."
          end
          exit 255
        end
      }
    end
  end

  
Lambda.start(ARGV)