Skip to content

E
ebulk
Commit 15ea30c8 authored by Roque's avatar Roque
Browse files
Options

Revert "[WIP] http requests are done using curl command" 

This reverts commit 469eb486.
http requests issue (see #1) solved
parent 469eb486
Hide whitespace changes
Inline Side-by-side
Showing with 77 additions and 94 deletions
ebulk-data/embulk-wendelin-dataset-tool/lib/embulk/wendelin_client.rb
View file @ 15ea30c8
......@@ -4,7 +4,6 @@ require 'openssl'
require 'yaml'
require 'open-uri'
require_relative 'filelogger'
require 'open3'
# class representing a Wendelin client
class WendelinClient
......@@ -25,37 +24,6 @@ class WendelinClient
@banned_references_list = []
@logger = LogManager.instance()
@last_ingestion = Time.new - 2
begin
# check credentials
uri = URI(URI.escape("#{@erp5_url}"))
uri = URI(URI.escape("#{uri.scheme}://#{uri.host}/erp5/getDataStreamList?data_set_reference="))
scheme = uri.scheme == 'https' ? "https://" : "http://"
check_credential_url = uri.to_s.sub! scheme, scheme + @user + ":" + @password + "@"
command = "curl --request GET '" + check_credential_url + "'"
stdout,stderr,status = Open3.capture3(command)
@logger.info("Checking credentials...")
print "[INFO] Checking credentials..."
if status.success?
if stdout.include? "'"
stdout = stdout.gsub! "'", '"'
end
dict = JSON.parse(stdout)
if dict["status_code"] == 1
@logger.error(dict["error_message"], print=TRUE)
@logger.abortExecution()
end
@logger.info("OK")
puts "OK"
else
@logger.error(HTTP_MESSAGE_400, print=TRUE)
@logger.error(stderr)
@logger.abortExecution()
end
rescue Exception => e
@logger.error("HTTP ERROR: " + e.to_s, print=TRUE)
@logger.error(e.backtrace)
@logger.abortExecution()
end
end
def checkReferenceChars(reference)
......@@ -68,9 +36,7 @@ class WendelinClient
checkReferenceChars(reference)
uri = URI(URI.escape("#{@erp5_url}/ingestionReferenceExists?reference=#{reference}"))
begin
response = handleRequest(uri, nil, nil, non_json_content=true)
res = response["message"] if response["success"] == TRUE
raise response["message"] if response["success"] == FALSE
res = open(uri, http_basic_authentication: [@user, @password]).read
rescue Exception => e
@logger.error("An error occurred while checking if reference exists: " + e.to_s)
@logger.error(e.backtrace)
......@@ -87,7 +53,7 @@ class WendelinClient
@logger.info("Deletion requested for reference #{reference}", print=TRUE)
checkReferenceChars(reference)
uri = URI(URI.escape("#{@erp5_url}/ERP5Site_invalidateIngestionObjects?reference=#{reference}"))
res = handleRequest(uri, nil, nil, non_json_content=true)
res = handleRequest(uri)
if res["success"] == FALSE
@logger.abortExecution()
end
......@@ -99,7 +65,7 @@ class WendelinClient
checkReferenceChars(reference)
checkReferenceChars(new_reference)
uri = URI(URI.escape("#{@erp5_url}/ERP5Site_renameIngestion?reference=#{reference}&new_reference=#{new_reference}"))
res = handleRequest(uri, nil, nil, non_json_content=true)
res = handleRequest(uri)
if res["success"] == FALSE
@logger.abortExecution()
end
......@@ -112,9 +78,8 @@ class WendelinClient
else
@logger.info("Increasing dataset version")
begin
uri = URI(URI.escape("#{@erp5_url}/ERP5Site_increaseDatasetVersion?reference=#{reference}"))
response = handleRequest(uri, nil, nil, non_json_content=true)
raise response["message"] if response["success"] == FALSE
uri = URI(URI.escape("#{@erp5_url}/ERP5Site_increaseDatasetVersion?reference=#{reference}"))
res = open(uri, http_basic_authentication: [@user, @password]).read
rescue Exception => e
@logger.error("An error occurred while increasing dataset version: " + e.to_s)
@logger.error(e.backtrace)
......@@ -185,27 +150,31 @@ class WendelinClient
success = FALSE
n_retry = 0
while ! success && n_retry < 10
response = handleRequest(uri, nil, nil, non_json_content=true)
if response["success"] == TRUE
chunk = response["message"]
if chunk.nil? || chunk.empty?
if first
yield chunk
end
@logger.info("Done", print=TRUE)
done = TRUE
else
first = FALSE
n_part += 1
yield chunk
end
success = TRUE
else
@logger.error("Error downloading data: " + response["message"], print=TRUE)
n_retry += 1
@logger.info("Retrying #{n_retry}/10...", print=TRUE)
sleep 30
end
begin
res = open(uri, http_basic_authentication: [@user, @password]) {
|content|
chunk = content.read()
if chunk.nil? || chunk.empty?
if first
yield chunk
end
@logger.info("Done", print=TRUE)
done = TRUE
else
first = FALSE
n_part += 1
yield chunk
end
}
success = TRUE
rescue Exception => e
exception = e
@logger.error("Error downloading data: " + e.to_s, print=TRUE)
@logger.error(e.backtrace)
n_retry += 1
@logger.info("Retrying #{n_retry}/10...", print=TRUE)
sleep 30
end
end
raise exception if not success
end
......@@ -226,41 +195,55 @@ class WendelinClient
end
private
def handleRequest(uri, reference=nil, data_chunk=nil, non_json_content=false)
scheme = uri.scheme == 'https' ? "https://" : "http://"
url = uri.to_s.sub! scheme, scheme + @user + ":" + @password + "@"
begin
command = "curl --request POST '" + url + "'"
def handleRequest(uri, reference=nil, data_chunk=nil)
req = Net::HTTP::Post.new(uri)
req.basic_auth @user, @password
if data_chunk != nil
command += " --data 'data_chunk=" + data_chunk + "'"
@logger.info("Setting request form data...", print=TRUE) if reference != nil
begin
req.set_form_data('data_chunk' => data_chunk)
rescue java.lang.OutOfMemoryError
@logger.logOutOfMemoryError(reference)
return {"success"=>FALSE, "message"=>HTTP_MEMORY_ERROR}
rescue Exception => e
@logger.error("Error setting form data: " + e.to_s, print=TRUE)
@logger.error(e.backtrace)
return {"success"=>FALSE, "message"=>HTTP_MESSAGE_EXCEPTION}
end
@logger.info("Sending record:'#{reference}'...", print=TRUE) if reference != nil
end
stdout,stderr,status = Open3.capture3(command)#) + " >&2")
if status.success?
# check response (if ok, it should be empty or json)
if not stdout.nil? and not stdout.empty?
begin
if not non_json_content
if stdout.include? "'"
stdout = stdout.gsub! "'", '"'
end
dict = JSON.parse(stdout)
end
rescue Exception => e
@logger.error("EXCEPTION: " + e.to_s, print=TRUE)
@logger.error(HTTP_MESSAGE_400, print=TRUE)
@logger.error(stdout)
end
end
@logger.info("Done")
return {"success"=>TRUE, "message"=>stdout}
begin
res = Net::HTTP.start(uri.hostname, uri.port,
:use_ssl => (uri.scheme == 'https'),
:verify_mode => OpenSSL::SSL::VERIFY_NONE,
:ssl_timeout => 300, :open_timeout => 300, :read_timeout => 300,
) do |http|
http.request(req)
end
rescue Exception => e
@logger.error("HTTP ERROR: " + e.to_s, print=TRUE)
@logger.error(e.backtrace)
return {"success"=>FALSE, "message"=>HTTP_MESSAGE_EXCEPTION}
else
@logger.error(HTTP_MESSAGE_OTHER, print=TRUE)
@logger.error(stderr)
end
rescue Exception => e
@logger.error("HTTP ERROR: " + e.to_s, print=TRUE)
@logger.error(e.backtrace)
return {"success"=>FALSE, "message"=>e.to_s}
end
if res.kind_of?(Net::HTTPSuccess) # res.code is 2XX
@logger.info("Done")
return {"success"=>TRUE, "message"=>res.body}
else
@logger.error("HTTP FAIL - code: #{res.code}", print=TRUE)
if res.code == '500' or res.code == '502' or res.code == '503'
@logger.error(HTTP_MESSAGE_5XX, print=TRUE)
elsif res.code == '401'
@logger.error(HTTP_MESSAGE_401, print=TRUE)
@logger.abortExecution()
elsif res.code == '400'
@logger.error(HTTP_MESSAGE_400, print=TRUE)
@logger.abortExecution()
else
@logger.error(HTTP_MESSAGE_OTHER, print=TRUE)
end
return {"success"=>FALSE, "message"=>HTTP_MESSAGE_NOT_2XX}
end
end
end
end
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7