QRunIO/qqq
2
0
Fork 0
mirror of https://github.com/Kingsrook/qqq.git synced 2025-07-17 20:50:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add basic rate limit handling for POST

Browse Source
This commit is contained in:
darin.kelkhoff
2022-10-18 11:00:28 -05:00
parent 6345846eba
commit 2d3d1091fd
4 changed files with 178 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
qqq-backend-module-api/pom.xml
View File

@ -34,6 +34,10 @@
<properties>
<!-- props specifically to this module -->
<!-- none at this time -->
<!-- todo - remove this once module is further built out and we can hit standard ratio -->
<coverage.instructionCoveredRatioMinimum>0.70</coverage.instructionCoveredRatioMinimum>
<coverage.classCoveredRatioMinimum>0.50</coverage.classCoveredRatioMinimum>
</properties>
<dependencies>
@ -77,6 +81,7 @@
<configuration>
<excludes>
<exclude>com/kingsrook/qqq/backend/module/api/model/**/*.class</exclude>
<exclude>com/kingsrook/qqq/backend/module/api/exceptions/**/*.class</exclude>
</excludes>
</configuration>
</plugin>

117
qqq-backend-module-api/src/main/java/com/kingsrook/qqq/backend/module/api/actions/APIInsertAction.java
View File

@ -23,6 +23,7 @@ package com.kingsrook.qqq.backend.module.api.actions;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import com.kingsrook.qqq.backend.core.actions.interfaces.InsertInterface;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
@ -30,10 +31,15 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertOutput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
import com.kingsrook.qqq.backend.core.utils.SleepUtils;
import com.kingsrook.qqq.backend.module.api.exceptions.RateLimitException;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.conn.HttpClientConnectionManager;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@ -66,33 +72,20 @@ public class APIInsertAction extends AbstractAPIAction implements InsertInterfac
preAction(insertInput);
HttpClientConnectionManager connectionManager = null;
try
{
HttpClientBuilder httpClientBuilder = HttpClientBuilder.create();
HttpClient client = httpClientBuilder.build();
String url = apiActionUtil.buildTableUrl(table);
HttpPost request = new HttpPost(url);
apiActionUtil.setupAuthorizationInRequest(request);
apiActionUtil.setupContentTypeInRequest(request);
apiActionUtil.setupAdditionalHeaders(request);
connectionManager = new PoolingHttpClientConnectionManager();
// todo - supports bulk post?
for(QRecord record : insertInput.getRecords())
{
try
{
request.setEntity(apiActionUtil.recordToEntity(table, record));
postOneRecord(insertOutput, table, connectionManager, record);
HttpResponse response = client.execute(request);
QRecord outputRecord = apiActionUtil.processPostResponse(table, record, response);
insertOutput.addRecord(outputRecord);
}
catch(Exception e)
if(insertInput.getRecords().size() > 1 && apiActionUtil.getMillisToSleepAfterEveryCall() > 0)
{
record.addError("Error: " + e.getMessage());
SleepUtils.sleep(apiActionUtil.getMillisToSleepAfterEveryCall(), TimeUnit.MILLISECONDS);
}
}
@ -100,9 +93,91 @@ public class APIInsertAction extends AbstractAPIAction implements InsertInterfac
}
catch(Exception e)
{
LOG.warn("Error in API Insert", e);
LOG.warn("Error in API Insert for [" + table.getName() + "]", e);
throw new QException("Error executing insert: " + e.getMessage(), e);
}
finally
{
if(connectionManager != null)
{
connectionManager.shutdown();
}
}
}
/*******************************************************************************
**
*******************************************************************************/
private void postOneRecord(InsertOutput insertOutput, QTableMetaData table, HttpClientConnectionManager connectionManager, QRecord record) throws RateLimitException
{
int sleepMillis = apiActionUtil.getInitialRateLimitBackoffMillis();
int rateLimitsCaught = 0;
while(true)
{
try
{
postOneTime(insertOutput, table, connectionManager, record);
return;
}
catch(RateLimitException rle)
{
rateLimitsCaught++;
if(rateLimitsCaught > apiActionUtil.getMaxAllowedRateLimitErrors())
{
LOG.warn("Giving up POST to [" + table.getName() + "] after too many rate-limit errors (" + apiActionUtil.getMaxAllowedRateLimitErrors() + ")");
record.addError("Error: " + rle.getMessage());
return;
}
LOG.info("Caught RateLimitException [#" + rateLimitsCaught + "] POST'ing to [" + table.getName() + "] - sleeping [" + sleepMillis + "]...");
SleepUtils.sleep(sleepMillis, TimeUnit.MILLISECONDS);
sleepMillis *= 2;
}
}
}
/*******************************************************************************
**
*******************************************************************************/
private void postOneTime(InsertOutput insertOutput, QTableMetaData table, HttpClientConnectionManager connectionManager, QRecord record) throws RateLimitException
{
try
{
CloseableHttpClient client = HttpClients.custom().setConnectionManager(connectionManager).build();
String url = apiActionUtil.buildTableUrl(table);
HttpPost request = new HttpPost(url);
apiActionUtil.setupAuthorizationInRequest(request);
apiActionUtil.setupContentTypeInRequest(request);
apiActionUtil.setupAdditionalHeaders(request);
request.setEntity(apiActionUtil.recordToEntity(table, record));
HttpResponse response = client.execute(request);
int statusCode = response.getStatusLine().getStatusCode();
if(statusCode == 429)
{
throw (new RateLimitException(EntityUtils.toString(response.getEntity())));
}
QRecord outputRecord = apiActionUtil.processPostResponse(table, record, response);
insertOutput.addRecord(outputRecord);
}
catch(RateLimitException rle)
{
throw (rle);
}
catch(Exception e)
{
LOG.warn("Error posting to [" + table.getName() + "]", e);
record.addError("Error: " + e.getMessage());
insertOutput.addRecord(record);
}
}
}

42
qqq-backend-module-api/src/main/java/com/kingsrook/qqq/backend/module/api/actions/BaseAPIActionUtil.java
View File

@ -41,6 +41,8 @@ import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.entity.AbstractHttpEntity;
import org.apache.http.entity.StringEntity;
import org.apache.http.util.EntityUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.json.JSONObject;
@ -50,11 +52,43 @@ import org.json.JSONObject;
*******************************************************************************/
public class BaseAPIActionUtil
{
private static final Logger LOG = LogManager.getLogger(BaseAPIActionUtil.class);
protected APIBackendMetaData backendMetaData;
protected AbstractTableActionInput actionInput;
/*******************************************************************************
**
*******************************************************************************/
public long getMillisToSleepAfterEveryCall()
{
return 0;
}
/*******************************************************************************
**
*******************************************************************************/
public int getInitialRateLimitBackoffMillis()
{
return 0;
}
/*******************************************************************************
**
*******************************************************************************/
public int getMaxAllowedRateLimitErrors()
{
return 0;
}
/*******************************************************************************
** As part of making a request - set up its authorization header (not just
** strictly "Authorization", but whatever is needed for auth).
@ -146,8 +180,8 @@ public class BaseAPIActionUtil
protected AbstractHttpEntity recordToEntity(QTableMetaData table, QRecord record) throws IOException
{
JSONObject body = recordToJsonObject(table, record);
String json = body.toString(3);
System.out.println(json);
String json = body.toString();
LOG.debug(json);
return (new StringEntity(json));
}
@ -191,11 +225,11 @@ public class BaseAPIActionUtil
protected QRecord processPostResponse(QTableMetaData table, QRecord record, HttpResponse response) throws IOException
{
int statusCode = response.getStatusLine().getStatusCode();
System.out.println(statusCode);
LOG.debug(statusCode);
HttpEntity entity = response.getEntity();
String resultString = EntityUtils.toString(entity);
System.out.println(resultString);
LOG.debug(resultString);
JSONObject jsonObject = JsonUtils.toJSONObject(resultString);

39
qqq-backend-module-api/src/main/java/com/kingsrook/qqq/backend/module/api/exceptions/RateLimitException.java Normal file
View File

@ -0,0 +1,39 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2022. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.kingsrook.qqq.backend.module.api.exceptions;
/*******************************************************************************
**
*******************************************************************************/
public class RateLimitException extends Exception
{
/*******************************************************************************
**
*******************************************************************************/
public RateLimitException(String message)
{
super(message);
}
}