You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1123 lines
38 KiB
1123 lines
38 KiB
require 'sequel'
|
|
require 'fileutils'
|
|
require_relative './user'
|
|
require_relative './table'
|
|
require_relative './visualization/member'
|
|
require_relative './table_registrar'
|
|
require_relative './quota_checker'
|
|
require_relative '../../lib/cartodb/errors'
|
|
require_relative '../../lib/cartodb/import_error_codes'
|
|
require_relative '../../lib/cartodb/metrics'
|
|
require_relative '../../lib/cartodb/stats/importer'
|
|
require_relative '../../config/initializers/redis'
|
|
require_relative '../../services/importer/lib/importer'
|
|
require_relative '../connectors/importer'
|
|
require_relative '../../services/importer/lib/importer/datasource_downloader'
|
|
require_relative '../../services/datasources/lib/datasources'
|
|
require_relative '../../services/importer/lib/importer/unp'
|
|
require_relative '../../services/importer/lib/importer/post_import_handler'
|
|
require_relative '../../services/importer/lib/importer/mail_notifier'
|
|
require_relative '../../services/importer/lib/importer/cartodbfy_time'
|
|
require_relative '../../services/platform-limits/platform_limits'
|
|
require_relative '../../services/importer/lib/importer/overviews'
|
|
require_relative '../../services/importer/lib/importer/connector_runner'
|
|
require_relative '../../services/importer/lib/importer/exceptions'
|
|
require_relative '../../services/importer/lib/importer/column'
|
|
|
|
require_dependency 'carto/tracking/events'
|
|
require_dependency 'carto/valid_table_name_proposer'
|
|
require_dependency 'carto/configuration'
|
|
require_dependency 'carto/db/user_schema'
|
|
require_dependency 'carto/uuidhelper'
|
|
require_dependency 'carto/helpers/data_import_commons'
|
|
|
|
include CartoDB::Datasources
|
|
|
|
class DataImport < Sequel::Model
|
|
include Carto::DataImportConstants
|
|
include Carto::Configuration
|
|
include Carto::UUIDHelper
|
|
include Carto::DataImportCommons
|
|
|
|
MERGE_WITH_UNMATCHING_COLUMN_TYPES_RE = /No .*matches.*argument type.*/
|
|
DIRECT_STATEMENT_TIMEOUT = 1.hour * 1000
|
|
|
|
attr_accessor :log, :results
|
|
|
|
many_to_one :user
|
|
|
|
# Not all constants are used, but so that we keep track of available states
|
|
STATE_ENQUEUED = 'enqueued' # Default state for imports whose files are not yet at "import source"
|
|
STATE_PENDING = 'pending' # Default state for files already at "import source" (e.g. S3 bucket)
|
|
STATE_UNPACKING = 'unpacking'
|
|
STATE_IMPORTING = 'importing'
|
|
STATE_COMPLETE = 'complete'
|
|
STATE_UPLOADING = 'uploading'
|
|
STATE_FAILURE = 'failure'
|
|
STATE_STUCK = 'stuck'
|
|
|
|
TYPE_EXTERNAL_TABLE = 'external_table'
|
|
TYPE_FILE = 'file'
|
|
TYPE_URL = 'url'
|
|
TYPE_QUERY = 'query'
|
|
TYPE_DATASOURCE = 'datasource'
|
|
|
|
def after_initialize
|
|
instantiate_log
|
|
self.results = []
|
|
self.state ||= STATE_PENDING
|
|
end
|
|
|
|
# This before_create should be only necessary to track old dashboard data imports.
|
|
# New ones are already tracked during the data_import create inside the controller
|
|
# For the old dashboard
|
|
def before_create
|
|
if from_common_data?
|
|
self.extra_options = extra_options.merge(common_data: true)
|
|
end
|
|
self.extra_options = extra_options.merge(column_sanitization_version: CartoDB::Importer2::Column::CURRENT_COLUMN_SANITIZATION_VERSION)
|
|
end
|
|
|
|
def before_save
|
|
unless logger.present?
|
|
log.save
|
|
self.logger = log.id
|
|
end
|
|
self.updated_at = Time.now
|
|
end
|
|
|
|
# The objective of this after_create method is to track in the logs every started
|
|
# import process.
|
|
def after_create
|
|
notify(results)
|
|
end
|
|
|
|
def from_common_data?
|
|
username = Cartodb.get_config(:common_data, 'username')
|
|
host = Cartodb.get_config(:common_data, 'host')
|
|
if username && host &&
|
|
!extra_options.has_key?('common_data') &&
|
|
data_source && data_source.include?("#{username}.#{host}")
|
|
return true
|
|
end
|
|
|
|
false
|
|
end
|
|
|
|
def extra_options
|
|
return {} if self.import_extra_options.nil?
|
|
::JSON.parse(self.import_extra_options).symbolize_keys
|
|
end
|
|
|
|
def extra_options=(value)
|
|
if !value.nil?
|
|
self.import_extra_options = ::JSON.dump(value)
|
|
end
|
|
end
|
|
|
|
def dataimport_logger
|
|
@@dataimport_logger ||= CartoDB.unformatted_logger(log_file_path("imports.log"))
|
|
end
|
|
|
|
def run_import!
|
|
self.resque_ppid = Process.ppid
|
|
self.server = Socket.gethostname
|
|
log.append("Running on server #{server} with PID: #{Process.pid}")
|
|
|
|
begin
|
|
self.state = STATE_UPLOADING
|
|
|
|
success = if table_copy.present? || from_query.present?
|
|
from_table
|
|
elsif service_name == 'connector'
|
|
!!execute_importer(*new_importer_with_connector)
|
|
else
|
|
!!execute_importer(*new_importer)
|
|
end
|
|
rescue TokenExpiredOrInvalidError => ex
|
|
success = false
|
|
begin
|
|
current_user.oauths.remove(ex.service_name)
|
|
rescue StandardError => ex2
|
|
log.append("Exception removing OAuth: #{ex2.message}")
|
|
log.append(ex2.backtrace)
|
|
end
|
|
end
|
|
|
|
log.append('After dispatch')
|
|
|
|
if results.empty?
|
|
if collision_strategy == COLLISION_STRATEGY_SKIP
|
|
self.error_code = 1022
|
|
self.state = STATE_COMPLETE
|
|
else
|
|
self.error_code = 1002
|
|
self.state = STATE_FAILURE
|
|
end
|
|
save
|
|
end
|
|
|
|
self.cartodbfy_time = CartoDB::Importer2::CartodbfyTime::instance(id).get
|
|
success ? handle_success : handle_failure
|
|
log.store
|
|
Rails.logger.debug log.to_s
|
|
|
|
self
|
|
rescue CartoDB::QuotaExceeded => quota_exception
|
|
current_user_id = current_user.id
|
|
Carto::Tracking::Events::ExceededQuota.new(current_user_id, user_id: current_user_id).report
|
|
|
|
CartoDB::notify_warning_exception(quota_exception)
|
|
handle_failure(quota_exception)
|
|
self
|
|
rescue CartoDB::CartoDBfyInvalidID
|
|
invalid_cartodb_id_exception = CartoDB::Importer2::CartoDBfyInvalidID.new
|
|
log.append("Exception: #{invalid_cartodb_id_exception}")
|
|
CartoDB::notify_warning_exception(invalid_cartodb_id_exception)
|
|
handle_failure(invalid_cartodb_id_exception)
|
|
self
|
|
rescue Carto::UnauthorizedError => e
|
|
log.append("Exception: #{e.message}")
|
|
log.append(e.backtrace, false)
|
|
stacktrace = e.message + e.backtrace.join
|
|
CartoDB.report_exception(e, 'Map quota exceeded', error_info: stacktrace)
|
|
error = CartoDB::Importer2::MapQuotaExceededError.new
|
|
handle_failure(error)
|
|
raise error
|
|
rescue StandardError => exception
|
|
log.append("Exception: #{exception.to_s}")
|
|
log.append(exception.backtrace, false)
|
|
stacktrace = exception.to_s + exception.backtrace.join
|
|
CartoDB.report_exception(exception, 'Import error', error_info: stacktrace)
|
|
handle_failure(exception)
|
|
raise exception
|
|
self
|
|
end
|
|
|
|
def get_error_source
|
|
if self.error_code == CartoDB::NO_ERROR_CODE
|
|
CartoDB::NO_ERROR_CODE
|
|
else
|
|
self.error_code.blank? ? CartoDB::IMPORTER_ERROR_CODES[99999][:source] : CartoDB::IMPORTER_ERROR_CODES[self.error_code][:source]
|
|
end
|
|
end
|
|
|
|
def raise_over_table_quota_error
|
|
log.append('Over account table limit, please upgrade')
|
|
self.error_code = 8002
|
|
self.state = STATE_FAILURE
|
|
save
|
|
raise CartoDB::QuotaExceeded, 'More tables required'
|
|
end
|
|
|
|
# TODO: move to new model
|
|
def mark_as_failed_if_stuck!
|
|
return false unless stuck?
|
|
|
|
log.append("Import timed out. Id:#{self.id} State:#{self.state} Created at:#{self.created_at} Running imports:#{running_import_ids}")
|
|
|
|
handle_failure(CartoDB::Importer2::StuckImportJobError.new)
|
|
|
|
CartoDB::notify_exception(
|
|
CartoDB::Importer2::GenericImportError.new('Import timed out or got stuck'),
|
|
user: current_user
|
|
)
|
|
true
|
|
end
|
|
|
|
def data_source=(data_source)
|
|
if data_source.nil?
|
|
values[:data_type] = TYPE_DATASOURCE
|
|
values[:data_source] = ''
|
|
elsif self.data_type != TYPE_QUERY
|
|
begin
|
|
path = uploaded_file_path(data_source)
|
|
if File.exist?(path) && !File.directory?(path)
|
|
values[:data_type] = TYPE_FILE
|
|
values[:data_source] = path
|
|
elsif Addressable::URI.parse(data_source).host.present?
|
|
values[:data_type] = TYPE_URL
|
|
values[:data_source] = data_source
|
|
end
|
|
rescue Addressable::URI::InvalidURIError
|
|
# this should only happen in testing, but just in case capture and log
|
|
log_warning(message: 'InvalidURIError when processing data_source', data_source: data_source)
|
|
end
|
|
end
|
|
|
|
self.original_url = self.values[:data_source] if (self.original_url.to_s.length == 0)
|
|
|
|
# else SQL-based import
|
|
end
|
|
|
|
def remove_uploaded_resources
|
|
return nil unless uploaded_file
|
|
|
|
file_upload_helper = CartoDB::FileUpload.new(Cartodb.get_config(:importer, 'uploads_path'))
|
|
path = file_upload_helper.get_uploads_path.join(uploaded_file[1])
|
|
FileUtils.rm_rf(path) if Dir.exists?(path)
|
|
end
|
|
|
|
def handle_success
|
|
self.success = true
|
|
self.state = STATE_COMPLETE
|
|
table_names = results.map { |result| result.name }.select { |name| name != nil}.sort
|
|
self.table_names = table_names.join(' ')
|
|
self.tables_created_count = table_names.size
|
|
log.append("Import finished\n")
|
|
log.store
|
|
save
|
|
|
|
begin
|
|
CartoDB::PlatformLimits::Importer::UserConcurrentImportsAmount.new({
|
|
user: current_user,
|
|
redis: {
|
|
db: $users_metadata
|
|
}
|
|
})
|
|
.decrement!
|
|
rescue StandardError => exception
|
|
log_info('Error decreasing concurrent import limit', exception: exception)
|
|
end
|
|
notify(results, id)
|
|
|
|
self
|
|
end
|
|
|
|
def handle_failure(supplied_exception = nil)
|
|
self.success = false
|
|
self.state = STATE_FAILURE
|
|
if !supplied_exception.nil? && supplied_exception.respond_to?(:error_code)
|
|
self.error_code = supplied_exception.error_code
|
|
end
|
|
log.append("ERROR!\n")
|
|
log.store
|
|
self.save
|
|
begin
|
|
CartoDB::PlatformLimits::Importer::UserConcurrentImportsAmount.new({
|
|
user: current_user,
|
|
redis: {
|
|
db: $users_metadata
|
|
}
|
|
})
|
|
.decrement!
|
|
rescue StandardError => exception
|
|
log_info('Error decreasing concurrent import limit', exception: exception)
|
|
end
|
|
notify(results)
|
|
self
|
|
rescue StandardError => exception
|
|
log.append("Exception: #{exception.to_s}")
|
|
log.append(exception.backtrace, false)
|
|
log.store
|
|
self
|
|
end
|
|
|
|
# Calculates the maximum timeout in seconds for a given user, to be used when performing HTTP requests
|
|
# TODO: Candidate for being private if we join syncs and data imports someday
|
|
# TODO: Add timeout config (if we need to change this)
|
|
def self.http_timeout_for(user, assumed_kb_sec = 75*1024)
|
|
if user.nil? || !user.respond_to?(:quota_in_bytes)
|
|
raise ArgumentError.new('Need a User object to calculate its download speed')
|
|
end
|
|
|
|
if assumed_kb_sec < 1
|
|
raise ArgumentError.new('KB per second must be > 0')
|
|
end
|
|
|
|
(user.quota_in_bytes / assumed_kb_sec).round
|
|
end
|
|
|
|
def validate
|
|
super
|
|
errors.add(:user, "Viewer users can't create data imports") if user && user.viewer
|
|
validate_collision_strategy
|
|
end
|
|
|
|
def final_state?
|
|
[STATE_COMPLETE, STATE_FAILURE, STATE_STUCK].include?(state)
|
|
end
|
|
|
|
def column_sanitization_version
|
|
extra_options[:column_sanitization_version] || CartoDB::Importer2::Column::INITIAL_COLUMN_SANITIZATION_VERSION
|
|
end
|
|
|
|
private
|
|
|
|
def get_provider_name_from_id(service_item_id)
|
|
begin
|
|
connector_params = JSON.parse(service_item_id)
|
|
return connector_params['provider']
|
|
rescue StandardError
|
|
return nil
|
|
end
|
|
end
|
|
|
|
def running_import_ids
|
|
Resque::Worker.all.map do |worker|
|
|
next unless worker.job['queue'] == 'imports'
|
|
worker.job['payload']['args'].first['job_id'] rescue nil
|
|
end.compact
|
|
end
|
|
|
|
def public_url
|
|
return data_source unless uploaded_file
|
|
"https://#{current_user.username}.carto.com/#{uploaded_file[0]}"
|
|
end
|
|
|
|
def before_destroy
|
|
self.remove_uploaded_resources
|
|
end
|
|
|
|
def instantiate_log
|
|
uuid = logger
|
|
|
|
self.log = if uuid?(uuid)
|
|
Carto::Log.find(uuid.to_s)
|
|
else
|
|
Carto::Log.new_data_import(user_id)
|
|
end
|
|
end
|
|
|
|
def uploaded_file
|
|
data_source.to_s.match(/uploads\/([a-z0-9]{20})\/.*/)
|
|
end
|
|
|
|
# A stuck job should've started but not be finished, so it's state should not be complete nor failed, it should
|
|
# have been in the queue for more than 5 minutes and it shouldn't be currently processed by any active worker
|
|
def stuck?
|
|
state == STATE_STUCK ||
|
|
![STATE_ENQUEUED, STATE_PENDING, STATE_COMPLETE, STATE_FAILURE].include?(state) &&
|
|
created_at < 5.minutes.ago &&
|
|
!running_import_ids.include?(id)
|
|
end
|
|
|
|
def from_table
|
|
log.append('from_table()')
|
|
|
|
number_of_tables = 1
|
|
quota_checker = CartoDB::QuotaChecker.new(current_user)
|
|
if quota_checker.will_be_over_table_quota?(number_of_tables)
|
|
raise_over_table_quota_error
|
|
end
|
|
|
|
query = table_copy ? "SELECT * FROM #{table_copy}" : from_query
|
|
new_table_name = import_from_query(table_name, query)
|
|
return true unless new_table_name && !overwrite_strategy?
|
|
|
|
sanitize_columns(new_table_name)
|
|
|
|
self.update(table_names: new_table_name, service_name: nil)
|
|
migrate_existing(new_table_name)
|
|
|
|
self.results.push CartoDB::Importer2::Result.new(success: true, error: nil)
|
|
rescue Sequel::DatabaseError => exception
|
|
if exception.to_s =~ MERGE_WITH_UNMATCHING_COLUMN_TYPES_RE
|
|
set_merge_error(8004, exception.to_s)
|
|
else
|
|
set_merge_error(8003, exception.to_s)
|
|
end
|
|
false
|
|
end
|
|
|
|
def import_from_query(name, query)
|
|
log.append('import_from_query()')
|
|
|
|
self.data_type = TYPE_QUERY
|
|
self.data_source = query
|
|
save
|
|
|
|
taken_names = Carto::Db::UserSchema.new(current_user).table_names
|
|
|
|
if taken_names.include?(name) && collision_strategy == Carto::DataImportConstants::COLLISION_STRATEGY_SKIP
|
|
log.append("Table with name #{name} already exists. Skipping")
|
|
return
|
|
end
|
|
|
|
table_name = Carto::ValidTableNameProposer.new.propose_valid_table_name(name, taken_names: taken_names)
|
|
|
|
if overwrite_strategy?
|
|
overwrite_table_from_query(table_name, name, query)
|
|
results.push CartoDB::Importer2::Result.new(success: true, error: nil)
|
|
else
|
|
current_user.db_service.in_database_direct_connection(
|
|
statement_timeout: DIRECT_STATEMENT_TIMEOUT
|
|
) do |user_direct_conn|
|
|
user_direct_conn.run(%{CREATE TABLE #{table_name} AS #{query}})
|
|
end
|
|
end
|
|
if current_user.over_disk_quota?
|
|
log.append("Over storage quota. Dropping table #{table_name}")
|
|
current_user.in_database.run(%{DROP TABLE #{table_name}})
|
|
self.error_code = 8001
|
|
self.state = STATE_FAILURE
|
|
save
|
|
raise CartoDB::QuotaExceeded, 'More storage required'
|
|
end
|
|
|
|
table_name
|
|
end
|
|
|
|
def overwrite_table_from_query(new_table_name, overwrite_table_name, query)
|
|
importer = new_importer_with_unused_runner
|
|
importer.overwrite_register(
|
|
CartoDB::Importer2::Result.new(tables: [new_table_name]),
|
|
overwrite_table_name
|
|
) do |database, schema|
|
|
database.execute(%{CREATE TABLE #{new_table_name} AS #{query}})
|
|
importer.drop("\"#{schema}\".\"#{overwrite_table_name}\"")
|
|
database.execute(%{
|
|
ALTER TABLE "#{schema}"."#{new_table_name}" RENAME TO "#{overwrite_table_name}"
|
|
})
|
|
end
|
|
end
|
|
|
|
def overwrite_strategy?
|
|
collision_strategy == Carto::DataImportConstants::COLLISION_STRATEGY_OVERWRITE
|
|
end
|
|
|
|
|
|
def sanitize_columns(table_name)
|
|
# TODO: is this called before table is registered? otherwise we should use Table#sanitize_columns
|
|
Table.sanitize_columns(
|
|
table_name,
|
|
column_sanitization_version,
|
|
{
|
|
connection: current_user.in_database,
|
|
database_schema: current_user.database_schema
|
|
}
|
|
)
|
|
end
|
|
|
|
def migrate_existing(imported_name)
|
|
log.append('migrate_existing()')
|
|
|
|
table = ::Table.new
|
|
table.user_id = user_id
|
|
table.name = imported_name
|
|
table.migrate_existing_table = imported_name
|
|
table.data_import_id = self.id
|
|
|
|
if table.valid?
|
|
log.append('Table valid')
|
|
table.save
|
|
table.optimize
|
|
table.map.recalculate_bounds!
|
|
if current_user.remaining_quota < 0
|
|
log.append('Over storage quota, removing table')
|
|
self.error_code = 8001
|
|
self.state = STATE_FAILURE
|
|
save
|
|
table.destroy
|
|
raise CartoDB::QuotaExceeded, 'More storage required'
|
|
end
|
|
refresh
|
|
self.table_id = table.id
|
|
self.table_name = table.name
|
|
log.append("Table '#{table.name}' registered")
|
|
save
|
|
true
|
|
else
|
|
reload
|
|
log.append("Table invalid: Error linking #{imported_name} to UI: " + table.errors.full_messages.join(' - '))
|
|
false
|
|
end
|
|
end
|
|
|
|
def pg_options
|
|
SequelRails.configuration.environment_for(Rails.env)
|
|
.merge(
|
|
username: current_user.database_username,
|
|
password: current_user.database_password,
|
|
database: current_user.database_name,
|
|
host: current_user.database_host
|
|
) {|key, o, n| n.nil? || n.empty? ? o : n}
|
|
end
|
|
|
|
def ogr2ogr_options
|
|
options = Cartodb.config.fetch(:ogr2ogr, {})
|
|
if options['binary'].nil? || options['csv_guessing'].nil?
|
|
{}
|
|
else
|
|
ogr_options = {
|
|
ogr2ogr_binary: options['binary'],
|
|
ogr2ogr_csv_guessing: options['csv_guessing'] && self.type_guessing,
|
|
quoted_fields_guessing: self.quoted_fields_guessing
|
|
}
|
|
if options['memory_limit'].present?
|
|
ogr_options.merge!(ogr2ogr_memory_limit: options['memory_limit'])
|
|
end
|
|
return ogr_options
|
|
end
|
|
end
|
|
|
|
def content_guessing_options
|
|
guessing_config = Cartodb.config.fetch(:importer, {}).deep_symbolize_keys.fetch(:content_guessing, {})
|
|
geocoder_config = Cartodb.config.fetch(:geocoder, {}).deep_symbolize_keys
|
|
if guessing_config[:enabled] and self.content_guessing and geocoder_config
|
|
{ guessing: guessing_config, geocoder: geocoder_config }
|
|
else
|
|
{ guessing: { enabled: false } }
|
|
end
|
|
end
|
|
|
|
# Create an Importer object (using a Runner to fetch the data).
|
|
# This methods returns an array with four elements:
|
|
# * importer: the new importer (nil if download errors detected)
|
|
# * runner: the runner that the importer uses
|
|
# * datasource_provider: the DataSource used
|
|
# * manual_fields: error code and log in case of errors
|
|
def new_importer
|
|
manual_fields = {}
|
|
had_errors = false
|
|
log.append('new_importer()')
|
|
|
|
datasource_provider = get_datasource_provider
|
|
|
|
# If retrieving metadata we get an error, fail early
|
|
begin
|
|
downloader = get_downloader(datasource_provider)
|
|
rescue DataDownloadError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1012,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue ExternalServiceError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1012,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue ExternalServiceTimeoutError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1020,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue DataDownloadTimeoutError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1020,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue ResponseError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1011,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue InvalidServiceError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1013,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue InvalidInputDataError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1012,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue UnsupportedOperationError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 1023,
|
|
log_info: ex.to_s
|
|
}
|
|
rescue CartoDB::Importer2::FileTooBigError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: ex.error_code,
|
|
log_info: CartoDB::IMPORTER_ERROR_CODES[ex.error_code]
|
|
}
|
|
rescue StandardError => ex
|
|
had_errors = true
|
|
manual_fields = {
|
|
error_code: 99999,
|
|
log_info: ex.to_s
|
|
}
|
|
end
|
|
|
|
if had_errors
|
|
importer = runner = datasource_provider = nil
|
|
else
|
|
post_import_handler = CartoDB::Importer2::PostImportHandler.new
|
|
case datasource_provider.class::DATASOURCE_NAME
|
|
when Url::ArcGIS::DATASOURCE_NAME
|
|
post_import_handler.add_fix_geometries_task
|
|
when Search::Twitter::DATASOURCE_NAME
|
|
post_import_handler.add_transform_geojson_geom_column
|
|
end
|
|
|
|
database_options = pg_options
|
|
self.host = database_options[:host]
|
|
|
|
unp = CartoDB::Importer2::Unp.new(Cartodb.config[:importer], Cartodb.config[:ogr2ogr])
|
|
|
|
importer, runner = new_importer_with_runner(downloader, unp, post_import_handler)
|
|
end
|
|
|
|
[importer, runner, datasource_provider, manual_fields]
|
|
end
|
|
|
|
# Create an Importer using a ConnectorRunner to fetch the data.
|
|
# This methods returns an array with two elements:
|
|
# * importer: the new importer (nil if download errors detected)
|
|
# * connector: the connector that the importer uses
|
|
def new_importer_with_connector
|
|
provider_name = get_provider_name_from_id(service_item_id)
|
|
CartoDB::Importer2::ConnectorRunner.check_availability!(current_user, provider_name)
|
|
|
|
database_options = pg_options
|
|
|
|
self.host = database_options[:host]
|
|
|
|
connector = CartoDB::Importer2::ConnectorRunner.new(
|
|
service_item_id,
|
|
user: current_user,
|
|
pg: database_options,
|
|
log: log,
|
|
collision_strategy: collision_strategy
|
|
)
|
|
|
|
registrar = CartoDB::TableRegistrar.new(current_user, ::Table)
|
|
quota_checker = CartoDB::QuotaChecker.new(current_user)
|
|
database = current_user.in_database
|
|
destination_schema = current_user.database_schema
|
|
public_user_roles = current_user.db_service.public_user_roles
|
|
overviews_creator = CartoDB::Importer2::Overviews.new(connector, current_user)
|
|
importer = CartoDB::Connector::Importer.new(
|
|
runner: connector,
|
|
table_registrar: registrar,
|
|
quota_checker: quota_checker,
|
|
database: database,
|
|
data_import_id: id,
|
|
overviews_creator: overviews_creator,
|
|
destination_schema: destination_schema,
|
|
public_user_roles: public_user_roles
|
|
)
|
|
[importer, connector, nil, nil]
|
|
end
|
|
|
|
# Create an Importer object (using a Runner to fetch the data).
|
|
# This methods returns an array with two elements:
|
|
# * importer: the new importer (nil if download errors detected)
|
|
# * runner: the runner that the importer uses
|
|
def new_importer_with_runner(downloader, unpacker, post_import_handler)
|
|
runner = CartoDB::Importer2::Runner.new(
|
|
pg: pg_options,
|
|
downloader: downloader,
|
|
log: log,
|
|
user: user,
|
|
unpacker: unpacker,
|
|
post_import_handler: post_import_handler,
|
|
importer_config: Cartodb.config[:importer],
|
|
collision_strategy: collision_strategy
|
|
)
|
|
|
|
runner.loader_options = ogr2ogr_options.merge content_guessing_options
|
|
runner.set_importer_stats_host_info(Socket.gethostname)
|
|
registrar = CartoDB::TableRegistrar.new(current_user, ::Table)
|
|
quota_checker = CartoDB::QuotaChecker.new(current_user)
|
|
database = current_user.in_database
|
|
destination_schema = current_user.database_schema
|
|
public_user_roles = current_user.db_service.public_user_roles
|
|
overviews_creator = CartoDB::Importer2::Overviews.new(runner, current_user)
|
|
importer = CartoDB::Connector::Importer.new(
|
|
runner: runner,
|
|
table_registrar: registrar,
|
|
quota_checker: quota_checker,
|
|
database: database,
|
|
data_import_id: id,
|
|
overviews_creator: overviews_creator,
|
|
destination_schema: destination_schema,
|
|
public_user_roles: public_user_roles,
|
|
collision_strategy: collision_strategy
|
|
)
|
|
|
|
[importer, runner]
|
|
end
|
|
|
|
# Create an Importer object with a runner that it's not able to fetch data.
|
|
# This method is useful when you just need some logic from the Importer class
|
|
# This methods returns an array with two elements:
|
|
# * importer: the new importer (nil if download errors detected)
|
|
def new_importer_with_unused_runner
|
|
importer, = new_importer_with_runner(nil, nil, nil)
|
|
importer
|
|
end
|
|
|
|
# Run importer, store results and return success state.
|
|
def execute_importer(importer, runner, datasource_provider = nil, manual_fields = nil)
|
|
if importer
|
|
tracker = lambda do |state|
|
|
self.state = state
|
|
save
|
|
end
|
|
log.append('Before importer run')
|
|
importer.run(tracker)
|
|
log.append('After importer run')
|
|
end
|
|
|
|
store_results(importer, runner, datasource_provider, manual_fields)
|
|
importer.nil? ? false : importer.success?
|
|
rescue StandardError => e
|
|
# Note: If this exception is not treated, results will not be defined
|
|
# and the import will finish with a null error_code
|
|
if manual_fields
|
|
set_error(manual_fields.fetch(:error_code, 99999))
|
|
end
|
|
raise e
|
|
end
|
|
|
|
# Note: Assumes that if importer is nil an error happened
|
|
# @param importer CartoDB::Connector::Importer|nil
|
|
# @param runner CartoDB::Importer2::Runner|nil
|
|
# @param datasource_provider mixed|nil
|
|
# @param manual_fields Hash
|
|
def store_results(importer=nil, runner=nil, datasource_provider=nil, manual_fields={})
|
|
if importer.nil?
|
|
set_error(manual_fields.fetch(:error_code, 99999), manual_fields.fetch(:log_info, nil))
|
|
else
|
|
self.results = importer.results
|
|
self.error_code = importer.error_code
|
|
self.rejected_layers = importer.rejected_layers.join(',') if !importer.rejected_layers.empty?
|
|
self.runner_warnings = runner.warnings.to_json if !runner.warnings.empty?
|
|
|
|
# http_response_code is only relevant if a direct download is performed
|
|
if runner && datasource_provider && datasource_provider.providers_download_url?
|
|
self.http_response_code = runner.downloader.http_response_code
|
|
end
|
|
|
|
# Table.after_create() setted fields that won't be saved to "final" data import unless specified here
|
|
self.table_name = importer.table.name if importer.success? && importer.table
|
|
self.table_id = importer.table.id if importer.success? && importer.table
|
|
|
|
if importer.success?
|
|
update_visualization_id(importer)
|
|
end
|
|
update_synchronization(importer)
|
|
|
|
importer.success? ? set_datasource_audit_to_complete(datasource_provider,
|
|
importer.success? && importer.table ? importer.table.id : nil)
|
|
: set_datasource_audit_to_failed(datasource_provider)
|
|
end
|
|
|
|
unless runner.nil?
|
|
self.stats = ::JSON.dump(runner.stats)
|
|
end
|
|
end
|
|
|
|
def update_visualization_id(importer)
|
|
if importer.data_import.create_visualization
|
|
self.visualization_id = importer.data_import.visualization_id
|
|
end
|
|
end
|
|
|
|
def update_synchronization(importer)
|
|
if synchronization_id
|
|
log.type = Carto::Log::TYPE_SYNCHRONIZATION
|
|
log.store
|
|
log.append("synchronization_id: #{synchronization_id}")
|
|
synchronization = CartoDB::Synchronization::Member.new(id: synchronization_id).fetch
|
|
synchronization.name = self.table_name
|
|
synchronization.log_id = log.id
|
|
|
|
if importer.success?
|
|
imported_table = ::Table.get_by_table_id(self.table_id)
|
|
if !imported_table.nil? && imported_table.table_visualization
|
|
synchronization.visualization_id = imported_table.table_visualization.id
|
|
end
|
|
|
|
synchronization.state = 'success'
|
|
synchronization.error_code = nil
|
|
synchronization.error_message = nil
|
|
synchronization.modified_at = importer.last_modified
|
|
else
|
|
synchronization.state = 'failure'
|
|
synchronization.error_code = error_code.blank? ? 9999 : error_code
|
|
synchronization.error_message = get_error_text[:title] + ' ' + get_error_text[:what_about]
|
|
end
|
|
log.append("importer.success? #{synchronization.state}")
|
|
synchronization.store
|
|
end
|
|
end
|
|
|
|
def get_datasource_provider
|
|
return nil if service_name == 'connector'
|
|
datasource_name = (service_name.nil? || service_name.size == 0) ? Url::PublicUrl::DATASOURCE_NAME : service_name
|
|
if service_item_id.nil? || service_item_id.size == 0
|
|
self.service_item_id = data_source
|
|
end
|
|
|
|
get_datasource(datasource_name, service_item_id)
|
|
end
|
|
|
|
def get_downloader(datasource_provider)
|
|
log.append("Fetching datasource #{datasource_provider} metadata for item id #{service_item_id}")
|
|
|
|
metadata = datasource_provider.get_resource_metadata(service_item_id)
|
|
|
|
if hit_platform_limit?(datasource_provider, metadata, current_user)
|
|
raise CartoDB::Importer2::FileTooBigError.new(metadata.inspect)
|
|
end
|
|
|
|
if datasource_provider.providers_download_url?
|
|
metadata_url = metadata[:url]
|
|
resource_url = (metadata_url.present? && datasource_provider.providers_download_url?) ? metadata_url : data_source
|
|
|
|
log.append("File will be downloaded from #{resource_url}")
|
|
|
|
http_options = { http_timeout: ::DataImport.http_timeout_for(current_user) }
|
|
CartoDB::Importer2::Downloader.new(current_user.id,
|
|
resource_url,
|
|
http_options,
|
|
importer_config: Cartodb.config[:importer])
|
|
else
|
|
log.append('Downloading file data from datasource')
|
|
|
|
http_timeout = ::DataImport.http_timeout_for(current_user)
|
|
options = {
|
|
http_timeout: http_timeout,
|
|
importer_config: Cartodb.config[:importer],
|
|
user_id: current_user.id
|
|
}
|
|
|
|
CartoDB::Importer2::DatasourceDownloader.new(datasource_provider, metadata, options, log)
|
|
end
|
|
end
|
|
|
|
def hit_platform_limit?(datasource, metadata, user)
|
|
if datasource.has_resource_size?(metadata)
|
|
CartoDB::PlatformLimits::Importer::InputFileSize.new({ user: user })
|
|
.is_over_limit!(metadata[:size])
|
|
else
|
|
false
|
|
end
|
|
end
|
|
|
|
def current_user
|
|
@current_user ||= ::User[user_id]
|
|
end
|
|
|
|
def notify(results, import_id=nil)
|
|
owner = ::User.where(:id => self.user_id).first
|
|
imported_tables = results.select {|r| r.success }.length
|
|
failed_tables = results.length - imported_tables
|
|
|
|
# Calculate total size out of stats
|
|
total_size = 0
|
|
::JSON.parse(stats).each { |stat| total_size += stat ? stat['size'] : 0 }
|
|
importer_stats_aggregator.update_counter('total_size', total_size)
|
|
|
|
import_time = self.updated_at - self.created_at
|
|
cartodbfy_throughtput = (cartodbfy_time == 0.0 ? nil : (total_size / cartodbfy_time))
|
|
|
|
import_log = {'user' => owner.username,
|
|
'state' => self.state,
|
|
'tables' => results.length,
|
|
'imported_tables' => imported_tables,
|
|
'failed_tables' => failed_tables,
|
|
'error_code' => self.error_code,
|
|
'import_timestamp' => Time.now,
|
|
'queue_server' => `hostname`.strip,
|
|
'database_host' => owner.database_host,
|
|
'service_name' => self.service_name,
|
|
'data_type' => self.data_type,
|
|
'is_sync_import' => sync?,
|
|
'import_time' => import_time,
|
|
'file_stats' => ::JSON.parse(self.stats),
|
|
'resque_ppid' => self.resque_ppid,
|
|
'user_timeout' => ::DataImport.http_timeout_for(current_user),
|
|
'error_source' => get_error_source,
|
|
'id' => self.id,
|
|
'total_size' => total_size,
|
|
'cartodbfy_time' => self.cartodbfy_time,
|
|
'import_throughput' => (total_size / import_time),
|
|
'cartodbfy_throughtput' => cartodbfy_throughtput,
|
|
'cartodbfy_import_ratio' => (self.cartodbfy_time / import_time)
|
|
}
|
|
if !self.extra_options.nil?
|
|
import_log['extra_options'] = self.extra_options
|
|
end
|
|
import_log.merge!(decorate_log(self))
|
|
dataimport_logger.info(import_log.to_json)
|
|
CartoDB::Importer2::MailNotifier.new(self, results, ::Resque).notify_if_needed
|
|
|
|
user_id = user.id
|
|
properties = {
|
|
user_id: user_id,
|
|
connection: {
|
|
imported_from: service_name,
|
|
data_from: data_type,
|
|
sync: sync?,
|
|
import_time: import_time,
|
|
data_size: total_size,
|
|
error_code: self.error_code
|
|
}
|
|
}
|
|
|
|
if service_name == 'connector'
|
|
connector_params = JSON.parse(service_item_id)
|
|
properties[:connection][:provider] = connector_params['provider']
|
|
end
|
|
|
|
if results.any?
|
|
results.each do |result|
|
|
CartoDB::Metrics.new.report(:import, payload_for(result))
|
|
|
|
properties[:connection][:file_type] = result.extension
|
|
|
|
if result.success?
|
|
Carto::Tracking::Events::CompletedConnection.new(user_id, properties).report
|
|
else
|
|
Carto::Tracking::Events::FailedConnection.new(user_id, properties).report
|
|
end
|
|
end
|
|
elsif state == STATE_FAILURE
|
|
Carto::Tracking::Events::FailedConnection.new(user_id, properties).report
|
|
end
|
|
|
|
track_results(results, import_id, properties) unless import_id.nil?
|
|
end
|
|
|
|
def importer_stats_aggregator
|
|
@importer_stats_aggregator ||= CartoDB::Stats::Importer.instance
|
|
end
|
|
|
|
def decorate_log(data_import)
|
|
decoration = { retrieved_items: 0 }
|
|
if data_import.success && data_import.table_id && data_import.from_query.nil? &&
|
|
data_import.table_copy.nil?
|
|
datasource = get_datasource_provider
|
|
if datasource && datasource.persists_state_via_data_import?
|
|
decoration = datasource.get_audit_stats
|
|
end
|
|
end
|
|
decoration
|
|
end
|
|
|
|
def payload_for(result=nil)
|
|
log.store
|
|
payload = {
|
|
file_url: public_url,
|
|
distinct_id: current_user.username,
|
|
username: current_user.username,
|
|
account_type: current_user.account_type,
|
|
database: current_user.database_name,
|
|
email: current_user.email,
|
|
log: log.to_s
|
|
}
|
|
payload.merge!(
|
|
name: result.name,
|
|
extension: result.extension,
|
|
success: result.success,
|
|
error_code: result.error_code,
|
|
) if result
|
|
payload.merge!(
|
|
file_url_hostname: URI.parse(public_url).hostname
|
|
) if public_url rescue nil
|
|
payload.merge!(error_title: get_error_text[:title]) if state == STATE_FAILURE
|
|
payload
|
|
end
|
|
|
|
# @param datasource_name String
|
|
# @param service_item_id String|nil
|
|
# @return mixed|nil
|
|
# @throws DataSourceError
|
|
def get_datasource(datasource_name, service_item_id)
|
|
begin
|
|
oauth = current_user.oauths.select(datasource_name)
|
|
# Tables metadata DB also store resque data
|
|
datasource = DatasourcesFactory.get_datasource(
|
|
datasource_name, current_user, {
|
|
http_timeout: ::DataImport.http_timeout_for(current_user),
|
|
redis_storage: $tables_metadata,
|
|
user_defined_limits: ::JSON.parse(user_defined_limits).symbolize_keys
|
|
})
|
|
datasource.token = oauth.token unless oauth.nil?
|
|
rescue StandardError => ex
|
|
log.append("Exception: #{ex.message}")
|
|
log.append(ex.backtrace, false)
|
|
CartoDB.report_exception(ex, 'Import error: ', error_info: ex.message + ex.backtrace.join)
|
|
raise CartoDB::DataSourceError.new("Datasource #{datasource_name} could not be instantiated")
|
|
end
|
|
if service_item_id.nil?
|
|
raise CartoDB::DataSourceError.new("Datasource #{datasource_name} without item id")
|
|
end
|
|
|
|
if datasource.persists_state_via_data_import?
|
|
datasource.data_import_item = self
|
|
end
|
|
|
|
datasource
|
|
end
|
|
|
|
def set_error(error_code, log_info='')
|
|
log.append("Additional error info: #{log_info}") unless log_info.empty?
|
|
self.results = [CartoDB::Importer2::Result.new(
|
|
success: false, error_code: error_code
|
|
)]
|
|
self.error_code = error_code
|
|
self.state = STATE_FAILURE
|
|
end
|
|
|
|
def set_merge_error(error_code, log_info='')
|
|
log.append("Going to set merge error with code #{error_code}")
|
|
set_error(error_code, log_info)
|
|
end
|
|
|
|
def set_datasource_audit_to_complete(datasource, table_id = nil)
|
|
if datasource && datasource.persists_state_via_data_import?
|
|
datasource.data_import_item = self
|
|
datasource.set_audit_to_completed(table_id)
|
|
end
|
|
end
|
|
|
|
def set_datasource_audit_to_failed(datasource)
|
|
if datasource && datasource.persists_state_via_data_import?
|
|
datasource.data_import_item = self
|
|
datasource.set_audit_to_failed
|
|
end
|
|
end
|
|
|
|
def track_results(results, import_id, import_properties)
|
|
current_user_id = current_user.id
|
|
return unless current_user_id
|
|
|
|
if visualization_id
|
|
Carto::Tracking::Events::CreatedMap.new(current_user_id, import_properties.merge(
|
|
user_id: current_user_id,
|
|
visualization_id: visualization_id,
|
|
origin: 'import')).report
|
|
end
|
|
|
|
results.select(&:success?).each do |result|
|
|
condition, origin = if result.name
|
|
[{ data_import_id: import_id, name: result.name },
|
|
from_common_data? ? 'common-data' : 'import']
|
|
else
|
|
[{ data_import_id: import_id }, 'copy']
|
|
end
|
|
|
|
user_table = ::UserTable.where(condition).first
|
|
map = user_table.map if user_table
|
|
if map
|
|
vis = Carto::Visualization.where(map_id: map.id).first
|
|
|
|
Carto::Tracking::Events::CreatedDataset.new(current_user_id, import_properties.merge(
|
|
user_id: current_user_id,
|
|
visualization_id: vis.id,
|
|
origin: origin)).report
|
|
end
|
|
end
|
|
rescue StandardError => e
|
|
log_warning(message: "Carto::Tracking: Couldn't report event", exception: e)
|
|
end
|
|
|
|
def sync?
|
|
synchronization_id.present?
|
|
end
|
|
end
|