Cron Job werfen DeadlineExceededError

Question

Ich arbeite derzeit an einem Google Cloud-Projekt im kostenlosen Testmodus. Ich habe cron Job, um die Daten von einem Datenanbieter zu holen und im Datenspeicher zu speichern. Ich schrieb den Code, um das Datenpaar von Wochen zu holen, und es war alles in Ordnung, aber plötzlich fing ich an, Fehler zu empfangen "DeadlineExceededError: Der Gesamttermin für die Antwort auf die HTTP-Anfrage wurde für die letzten zwei Tage überschritten". Ich glaube cron job soll nur nach 60 minuten aussetzen eine idee warum ich den fehler bekomme ?.

Crontask

def run(): 
    try: 
    config = cron.config 
    actual_data_source = config['xxx']['xxxx'] 
    original_data_source = actual_data_source 

    company_list = cron.rest_client.load(config, "companies", '') 

    if not company_list: 
     logging.info("Company list is empty") 
     return "Ok" 

    for row in company_list: 
     company_repository.save(row,original_data_source, actual_data_source) 

    return "OK" 

Repository Code

def save(dto, org_ds , act_dp): 
    try: 
    key = 'FIN/%s' % (dto['ticker']) 
    company = CompanyInfo(id=key) 
    company.stock_code = key 
    company.ticker = dto['ticker'] 
    company.name = dto['name'] 
    company.original_data_source = org_ds 
    company.actual_data_provider = act_dp 
    company.put() 
    return company 
    except Exception: 
    logging.exception("company_repository: error occurred saving the company 
    record ") 
    raise 

RestClient

def load(config, resource, filter): 
    try: 
    username = config['xxxx']['xxxx'] 
    password = config['xxxx']['xxxx'] 
    headers = {"Authorization": "Basic %s" % base64.b64encode(username + ":" 
    + password)} 

    if filter: 
     from_date = filter['from'] 
     to_date = filter['to'] 
     ticker = filter['ticker'] 
     start_date = datetime.strptime(from_date, '%Y%m%d').strftime("%Y-%m-%d") 
     end_date = datetime.strptime(to_date, '%Y%m%d').strftime("%Y-%m-%d") 

    current_page = 1 
    data = [] 

    while True: 

     if (filter): 
     url = config['xxxx']["endpoints"][resource] % (ticker, current_page, start_date, end_date) 
     else: 
     url = config['xxxx']["endpoints"][resource] % (current_page) 

     response = urlfetch.fetch(
      url=url, 
      deadline=60, 
      method=urlfetch.GET, 
      headers=headers, 
      follow_redirects=False, 

     ) 
     if response.status_code != 200: 
      logging.error("xxxx GET received status code %d!" % (response.status_code)) 
      logging.error("error happend for url: %s with headers %s", url, headers) 
      return 'Sorry, xxxx API request failed', 500 

     db = json.loads(response.content) 

     if not db['data']: 
      break 

     data.extend(db['data']) 

     if db['total_pages'] == current_page: 
      break 

     current_page += 1 

    return data 
except Exception: 
    logging.exception("Error occured with xxxx API request") 
    raise 

Answer 1

ich lieber zu schreibe dies als Kommentar, aber ich brauche mehr Reputation, um das zu tun.

1. Was passiert, wenn Sie den tatsächlichen Datenabruf direkt anstelle von über den Cron-Job ausführen?
2. Haben Sie versucht, eine Zeit Delta von Anfang bis Ende der Job zu messen?
3. Hat sich die Anzahl der abgerufenen Unternehmen dramatisch erhöht?
4. Sie scheinen eine Form der Aktien-Kurs-Aggregation zu tun - ist es möglich, dass der Provider hat begonnen, Sie zu blockieren?

Answer 2

Ich vermute, das die gleiche Frage wie dies, aber jetzt mit mehr Code: DeadlineExceededError: The overall deadline for responding to the HTTP request was exceeded

ich Ihren Code geändert in die Datenbank nach jedem urlfetch zu schreiben. Wenn mehr Seiten vorhanden sind, wird es in einer verzögerten Aufgabe neu gestartet, die weit vor dem 10-Minuten-Timeout liegen sollte.

Nicht abgefangene Ausnahmen in einer verzögerten Aufgabe führen dazu, dass es erneut versucht wird. Achten Sie also darauf.

Es war mir unklar, wie actual_data_source & original_data_source funktioniert, aber ich denke, dass Sie in der Lage sein sollten, diesen Teil zu ändern.

crontask

def run(current_page=0): 
    try: 
    config = cron.config 
    actual_data_source = config['xxx']['xxxx'] 
    original_data_source = actual_data_source 

    data, more = cron.rest_client.load(config, "companies", '', current_page) 

    for row in data: 
      company_repository.save(row, original_data_source, actual_data_source) 

    # fetch the rest 
    if more: 
     deferred.defer(run, current_page + 1) 
    except Exception as e: 
    logging.exception("run() experienced an error: %s" % e) 

RestClient

def load(config, resource, filter, current_page): 
    try: 
     username = config['xxxx']['xxxx'] 
     password = config['xxxx']['xxxx'] 
     headers = {"Authorization": "Basic %s" % base64.b64encode(username + ":" 
     + password)} 

     if filter: 
      from_date = filter['from'] 
      to_date = filter['to'] 
      ticker = filter['ticker'] 
      start_date = datetime.strptime(from_date, '%Y%m%d').strftime("%Y-%m-%d") 
      end_date = datetime.strptime(to_date, '%Y%m%d').strftime("%Y-%m-%d") 

      url = config['xxxx']["endpoints"][resource] % (ticker, current_page, start_date, end_date) 
     else: 
      url = config['xxxx']["endpoints"][resource] % (current_page) 

     response = urlfetch.fetch(
       url=url, 
       deadline=60, 
       method=urlfetch.GET, 
       headers=headers, 
       follow_redirects=False, 

     ) 
     if response.status_code != 200: 
       logging.error("xxxx GET received status code %d!" % (response.status_code)) 
       logging.error("error happend for url: %s with headers %s", url, headers) 
       return [], False 

     db = json.loads(response.content) 

     return db['data'], (db['total_pages'] != current_page) 


    except Exception as e: 
     logging.exception("Error occured with xxxx API request: %s" % e) 
     return [], False