require 'flydata-core/thread_context' require 'flydata-core/string_utils' module FlydataCore # Error 12xx are derived from AWS. # Error 100xx are FlyData errros. module ErrorCode VALUE_OUT_OF_RANGE_ERROR = 1204 INVALID_DATE_FORMAT_ERROR = 1205 INVALID_TIMESTAMP_FORMAT_ERROR = 1206 INVALID_NUMBER_ERROR = 1207 INVALID_FLOAT_ERROR = 1208 INVALID_DECIMAL_ERROR = 1209 INVALID_BOOLEAN_ERROR = 1210 NOT_NULL_VIOLATION_ERROR = 1213 INVALID_VARCHAR_ERROR = 1214 INVALID_CHAR_ERROR = 1215 INVALID_UTF8_CHARACTER_ERROR = 1220 INVALID_DATA_FORMAT_ERROR = 1201 MORE_COLUMNS_ERROR = 1202 LESS_COLUMNS_ERROR = 1203 NO_DATA_ERROR = 1211 RESERVED_TABLE_NAME_ERROR = 10001 RESERVED_COLUMN_NAME_ERROR = 10002 TABLE_MISSING_ERROR = 10003 AUTHENTICATION_ERROR = 10004 RETRY_LIMIT_EXCEED_ERROR = 10005 INFORMATION_SCHEMA_ERROR = 10006 TABLE_REVISION_ERROR = 10007 S3_ACCESS_ERROR = 10008 REDSHIFT_ACCESS_ERROR = 10009 BREAKING_ALTER_TABLE_ERROR = 10010 BREAKING_INFORMATION_SCHEMA_ERROR = 10011 BREAKING_TABLE_DDL_ERROR = 10012 INTERNAL_ERROR = 1200 end # Error Level # 0 Emergency: system is unusable # 1 Alert: action must be taken immediately # 2 Critical: critical conditions # 3 Error: error conditions # 4 Warning: warning conditions # 5 Notice: normal but significant condition # 6 Informational: informational messages # 7 Debug: debug-level messages module ErrorLevel ERROR = 3 # failed to load a record WARN = 4 # loaded a record but a record is modified end # RetryableError class RetryableError < StandardError NO_RETRY_LIMIT = :no_retry_limit def initialize(original_exception, retry_alert_limit = nil) @original_exception = original_exception @retry_alert_limit = retry_alert_limit end attr_reader :original_exception attr_reader :retry_alert_limit end # UserMaintenanceError class UserMaintenanceError < StandardError end # Errors usded by including module BreakingSyncError end # An error that indicates that sync record was received for a table whose sync is broken # These records cannot be retried, should be ignored class BrokenTableSyncError < StandardError end # FlyDataError # data_node_id, data_entry_id, chunk_identifier, table_name, err_code, err_reason, err_level class FlydataError < StandardError def initialize(error_reason, error_content = {}) error_content = if error_reason.kind_of?(String) { err_reason: error_reason}.merge(error_content) elsif error_reason.kind_of?(Hash) error_reason.merge(error_content) else raise ArgumentError.new( "Invalid argument. First parameter must be string or hash object.") end @error_content = default_params. merge(context_params). merge(error_content). merge(overwrite_params) filter_content(@error_content) @error_content[:err_code] = err_code end attr_reader :error_content def default_params if h = FlydataCore::ThreadContext.parameters {data_node_id: h[:node_id], data_entry_id: h[:data_entry_id], table_name: h[:table_name], chunk_identifier: h[:chunk_identifier], err_level: ErrorLevel::ERROR} else {} end end def context_params DataDeliveryErrorThreadContext.context_params || {} end def overwrite_params {} end def err_code raise NoMethodError, "default_parameters is not implemented" end def warn? err_level == ErrorLevel::WARN end def err_level @error_content[:err_level] end def to_s if r = @error_content[:err_reason] r else super end end MAX_CONTENT_SIZE = 1024 # 1kb def filter_content(error_content = @error_content) error_content.each do |k, value| if value.kind_of?(String) if value.length > MAX_CONTENT_SIZE value = error_content[k] = value[0..MAX_CONTENT_SIZE-1] end error_content[k] = FlydataCore::StringUtils.replace_invalid_utf8_char(value) end end end end # DataDeliveryError # An error which results in data clog or loss of information. # This type of error gets reported to customer as data delivery error. # data_node_id, data_entry_id, chunk_identifier, table_name, err_code, err_reason, err_level class DataDeliveryError < FlydataError end # + RecordDeliveryError # data_node_id, data_entry_id, chunk_identifier, table_name, err_code, err_reason, err_level, record_no, raw_record class RecordDeliveryError < DataDeliveryError def to_s "#{super} raw_record=(#{@error_content[:raw_record]})" end end class InvalidDataFormatError < RecordDeliveryError def err_code; ErrorCode::INVALID_DATA_FORMAT_ERROR; end end class MoreColumnsError < RecordDeliveryError def err_code; ErrorCode::MORE_COLUMNS_ERROR; end end class LessColumnsError < RecordDeliveryError def err_code; ErrorCode::LESS_COLUMNS_ERROR; end end class NoDataError < RecordDeliveryError def err_code; ErrorCode::NO_DATA_ERROR; end end class ReservedTableNameError < RecordDeliveryError def err_code; ErrorCode::RESERVED_TABLE_NAME_ERROR; end end class ReservedColumnNameError < RecordDeliveryError def err_code; ErrorCode::RESERVED_COLUMN_NAME_ERROR; end end class TableMissingError < RecordDeliveryError def err_code; ErrorCode::TABLE_MISSING_ERROR; end end class RecordDeliveryInternalError < RecordDeliveryError def err_code; ErrorCode::INTERNAL_ERROR; end end # information schema, especially flydata_ctl_columns table, is broken class InformationSchemaError < RecordDeliveryError def err_code; ErrorCode::INFORMATION_SCHEMA_ERROR; end end # No table information found in the database class MissingTableInformationError < InformationSchemaError def err_code; ErrorCode::INFORMATION_SCHEMA_ERROR; end end # Table's revision does not match the record's revision. class TableRevisionError < RecordDeliveryError def err_code; ErrorCode::TABLE_REVISION_ERROR; end end # Table's revision is lower than the record's revision. Should catch up soon. class StaleTableRevisionError < FlydataError def err_code; ErrorCode::TABLE_REVISION_ERROR; end end # Unsupported ALTER TABLE which breaks the table's sync consistency. class BreakingAlterTableError < RecordDeliveryError include BreakingSyncError def err_code; ErrorCode::BREAKING_ALTER_TABLE_ERROR; end end # Unsupported DDL which breaks the table's sync consistency. class BreakingTableDdlError < RecordDeliveryError include BreakingSyncError def err_code; ErrorCode::BREAKING_TABLE_DDL_ERROR; end end # Unrecoverable information schema error which requires re-synchronize a table class BreakingInformationSchemaError < InformationSchemaError include BreakingSyncError def initialize(*args) super @error_content[:err_reason] = "Sync is broken. The table needs to be re-synchronized. (#{@error_content[:err_reason]})" end def err_code; ErrorCode::BREAKING_INFORMATION_SCHEMA_ERROR; end end # + BadValueError # data_node_id, data_entry_id, chunk_identifier, table_name, err_code, err_reason, err_level, record_no, raw_record, colname, type, raw_value, modified_value class BadValueError < RecordDeliveryError def modified_value @error_content[:modified_value] end def has_modified_value? @error_content.has_key?(:modified_value) end def to_s "#{super} column=(#{@error_content[:col_name]}) value=(#{@error_content[:raw_value]})" end end class ValueOutOfRangeError < BadValueError def err_code; ErrorCode::VALUE_OUT_OF_RANGE_ERROR; end end class InvalidDateFormatError < BadValueError def err_code; ErrorCode::INVALID_DATE_FORMAT_ERROR; end end class InvalidTimestampFormatError < BadValueError def err_code; ErrorCode::INVALID_TIMESTAMP_FORMAT_ERROR; end end class InvalidNumberError < BadValueError def err_code; ErrorCode::INVALID_NUMBER_ERROR; end end class InvalidFloatError < BadValueError def err_code; ErrorCode::INVALID_FLOAT_ERROR; end end class InvalidDecimalError < BadValueError def err_code; ErrorCode::INVALID_DECIMAL_ERROR; end end class InvalidBooleanError < BadValueError def err_code; ErrorCode::INVALID_BOOLEAN_ERROR; end end class NotNullViolationError < BadValueError def err_code; ErrorCode::NOT_NULL_VIOLATION_ERROR; end end class InvalidVarcharError < BadValueError def err_code; ErrorCode::INVALID_VARCHAR_ERROR; end end class InvalidCharError < BadValueError def err_code; ErrorCode::INVALID_CHAR_ERROR; end end class InvalidUtf8CharacterError < BadValueError def err_code; ErrorCode::INVALID_UTF8_CHARACTER_ERROR; end end # + ChunkDeliveryError # data_node_id, data_entry_id, chunk_identifier, table_name, err_code, err_reason, err_level class ChunkDeliveryError < DataDeliveryError end class AuthenticationError < ChunkDeliveryError # end_point_url def err_code; ErrorCode::AUTHENTICATION_ERROR; end end class RetryLimitExceedError < ChunkDeliveryError def err_code; ErrorCode::RETRY_LIMIT_EXCEED_ERROR; end end class ChunkDeliveryInternalError < ChunkDeliveryError def err_code; ErrorCode::INTERNAL_ERROR; end end class UserResourceError < ChunkDeliveryError def default_params super.merge(timestamp: Time.now.utc) end def overwrite_params super.merge(table_name: nil) end end class S3AccessError < UserResourceError def err_code; ErrorCode::S3_ACCESS_ERROR; end end class RedshiftAccessError < UserResourceError def err_code; ErrorCode::REDSHIFT_ACCESS_ERROR; end end # error_content[:errors] has a list of errors class MultipleErrors < DataDeliveryError def err_code; -1; end end ## Depricated errors # The record is valid but not supported. class UnsupportedRecordFormat < StandardError; end # Table Def Error class TableDefError < StandardError attr_reader :err_hash def initialize(err_hash) super(err_hash[:error]) @err_hash = err_hash end end ## Compatibility check error class CompatibilityError < StandardError end class AgentCompatibilityError < CompatibilityError end class MysqlCompatibilityError < CompatibilityError end class PostgresqlCompatibilityError < CompatibilityError end ## Error container class DataDeliveryErrorThreadContext THREAD_LOCAL_KEY_ERROR_LIST = :flydata_data_delivery_error_list THREAD_LOCAL_KEY_RECORD_LIMIT = "#{THREAD_LOCAL_KEY_ERROR_LIST}_limit".to_sym DEFAULT_RECORD_LIMIT = 10 THREAD_LOCAL_KEY_CONTEXT_PARAMS = "#{THREAD_LOCAL_KEY_ERROR_LIST}_context_params".to_sym def self.initialize(record_limit = DEFAULT_RECORD_LIMIT) reset Thread.current[THREAD_LOCAL_KEY_RECORD_LIMIT] = record_limit end def self.add_error(error) return if error_list.include?(error) if error_list.count < record_limit error_list << error end #TODO: write warning log if record limit end def self.reset Thread.current[THREAD_LOCAL_KEY_ERROR_LIST] = nil Thread.current[THREAD_LOCAL_KEY_RECORD_LIMIT] = nil Thread.current[THREAD_LOCAL_KEY_CONTEXT_PARAMS] = nil end def self.record_limit Thread.current[THREAD_LOCAL_KEY_RECORD_LIMIT] || DEFAULT_RECORD_LIMIT end def self.error_list unless list = Thread.current[THREAD_LOCAL_KEY_ERROR_LIST] list = Thread.current[THREAD_LOCAL_KEY_ERROR_LIST] = [] end list end def self.set_context_params(params) Thread.current[THREAD_LOCAL_KEY_CONTEXT_PARAMS] = params end def self.add_context_params(params) cur_param = Thread.current[THREAD_LOCAL_KEY_CONTEXT_PARAMS] || {} Thread.current[THREAD_LOCAL_KEY_CONTEXT_PARAMS] = cur_param.merge(params) end def self.context_params Thread.current[THREAD_LOCAL_KEY_CONTEXT_PARAMS] end end end