Incremental loading of Github events and issues. #16
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,30 +8,38 @@ | |
| import org.elasticsearch.action.bulk.BulkProcessor; | ||
| import org.elasticsearch.action.bulk.BulkRequest; | ||
| import org.elasticsearch.action.bulk.BulkResponse; | ||
| import org.elasticsearch.action.index.IndexRequest; | ||
| import org.elasticsearch.action.search.SearchResponse; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.common.Base64; | ||
| import org.elasticsearch.common.inject.Inject; | ||
| import org.elasticsearch.common.settings.ImmutableSettings; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.xcontent.support.XContentMapValues; | ||
| import org.elasticsearch.index.query.FilterBuilders; | ||
| import org.elasticsearch.index.query.FilteredQueryBuilder; | ||
| import org.elasticsearch.index.query.QueryBuilders; | ||
| import org.elasticsearch.indices.IndexAlreadyExistsException; | ||
| import org.elasticsearch.river.AbstractRiverComponent; | ||
| import org.elasticsearch.river.River; | ||
| import org.elasticsearch.river.RiverName; | ||
| import org.elasticsearch.river.RiverSettings; | ||
| import org.elasticsearch.search.sort.FieldSortBuilder; | ||
| import org.elasticsearch.search.sort.SortBuilders; | ||
| import org.elasticsearch.search.sort.SortOrder; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.io.InputStreamReader; | ||
| import java.net.HttpURLConnection; | ||
| import java.net.URL; | ||
| import java.net.URLConnection; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.regex.Matcher; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery; | ||
| import static org.elasticsearch.index.query.QueryBuilders.termQuery; | ||
|
|
||
|
|
||
|
|
@@ -41,11 +49,14 @@ public class GitHubRiver extends AbstractRiverComponent implements River { | |
| private final String index; | ||
| private final String repository; | ||
| private final String owner; | ||
| private final int userRequestedInterval; | ||
| private final String endpoint; | ||
|
|
||
| private String password; | ||
| private String username; | ||
| private DataStream dataStream; | ||
| private String eventETag = null; | ||
| private int pollInterval = 60; | ||
|
|
||
| @SuppressWarnings({"unchecked"}) | ||
| @Inject | ||
|
|
@@ -61,8 +72,9 @@ public GitHubRiver(RiverName riverName, RiverSettings settings, Client client) { | |
| Map<String, Object> githubSettings = (Map<String, Object>) settings.settings().get("github"); | ||
| owner = XContentMapValues.nodeStringValue(githubSettings.get("owner"), null); | ||
| repository = XContentMapValues.nodeStringValue(githubSettings.get("repository"), null); | ||
|
|
||
| index = String.format("%s&%s", owner, repository); | ||
| userRequestedInterval = XContentMapValues.nodeIntegerValue(githubSettings.get("interval"), 60); | ||
|
|
||
| // auth (optional) | ||
| username = null; | ||
|
|
@@ -89,7 +101,7 @@ public void start() { | |
| client.admin().indices().prepareCreate(index).setSettings(indexSettings).execute().actionGet(); | ||
| logger.info("Created index."); | ||
| } catch (IndexAlreadyExistsException e) { | ||
| logger.info("Index already created"); | ||
| } catch (Exception e) { | ||
| logger.error("Exception creating index.", e); | ||
| } | ||
|
|
@@ -101,6 +113,7 @@ public void start() { | |
| @Override | ||
| public void close() { | ||
| dataStream.setRunning(false); | ||
| dataStream.interrupt(); | ||
| logger.info("Stopped GitHub river."); | ||
| } | ||
|
|
||
|
|
@@ -113,13 +126,38 @@ public DataStream() { | |
| isRunning = true; | ||
| } | ||
|
|
||
| private boolean checkAndUpdateETag(HttpURLConnection conn) throws IOException { | ||
| if (eventETag != null) { | ||
| conn.setRequestProperty("If-None-Match", eventETag); | ||
| } | ||
|
|
||
| String xPollInterval = conn.getHeaderField("X-Poll-Interval"); | ||
| if (xPollInterval != null) { | ||
| logger.debug("Next GitHub specified minimum polling interval is {} s", xPollInterval); | ||
| pollInterval = Integer.parseInt(xPollInterval); | ||
| } | ||
|
|
||
| if (conn.getResponseCode() == 304) { | ||
| logger.debug("304 {}", conn.getResponseMessage()); | ||
| return false; | ||
| } | ||
|
|
||
| String eTag = conn.getHeaderField("ETag"); | ||
| if (eTag != null) { | ||
| logger.debug("New eTag: {}", eTag); | ||
| eventETag = eTag; | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| private boolean indexResponse(HttpURLConnection conn, String type) { | ||
| InputStream input; | ||
| try { | ||
| input = conn.getInputStream(); | ||
| } catch (IOException e) { | ||
| logger.info("Exception encountered (403 usually is rate limit exceeded): ", e); | ||
| return false; | ||
| } | ||
| JsonStreamParser jsp = new JsonStreamParser(new InputStreamReader(input)); | ||
|
|
||
|
|
@@ -139,10 +177,17 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure) | |
| } | ||
| }).build(); | ||
|
|
||
| boolean continueIndexing = true; | ||
|
|
||
| IndexRequest req = null; | ||
| for (JsonElement e: array) { | ||
| if (type.equals("event")) { | ||
| req = indexEvent(e); | ||
| if (req == null) { | ||
| continueIndexing = false; | ||
| logger.debug("Found existing event, all remaining events has already been indexed"); | ||
|
There was a problem hiding this comment. Let's be 100% sure about this :D. Wouldn't want to miss any events! There was a problem hiding this comment. Yeah, that would be bad. We could run through them all in the beginning, just to be sure (though we lose quite a bit of the speedup). If we still make the check, we could log a warning (or at least an info) if an unindexed event is registered after the continueIndexing flag (which should probably be renamed in that case) has been set. If we see none like this for a few releases, it should be safe to enable the optimization. A middle ground could also be to run through the rest of the current block, and keep the continueIndexing warning. If there is any bugs or bad assumptions, there would still be a chance of losing events, but at least it would be quite a bit smaller, as it would have to hit in the boundary between blocks at for thing like off-by-one bugs to have an effect. If "last event first" doesn't hold in all cases, then all bets are of course off. It's actually a liiiitle strange that they didn't implement "since" for events. There was a problem hiding this comment. Why don't we try the first way you suggested for 1-2 days and watch the logs. If we see no occurrence, we should be fine. re: since for events - maybe there is a way. I suggest you shoot an email to support@github.com, I've done it a couple times and they were very helpful. Maybe there's something that we are missing? There was a problem hiding this comment. I sent them a mail, so let's see what they say. I'll make the change tonight. :) |
||
| break; | ||
| } | ||
| } else if (type.equals("issue")) { | ||
| req = indexOther(e, "IssueData", true); | ||
| } else if (type.equals("pullreq")) { | ||
|
|
@@ -160,13 +205,26 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure) | |
|
|
||
| try { | ||
| input.close(); | ||
| } catch (IOException e) { | ||
| logger.warn("Couldn't close connection?", e); | ||
| } | ||
|
|
||
| return continueIndexing; | ||
| } | ||
|
|
||
| private boolean isEventIndexed(String id) { | ||
| return client.prepareGet(index, null, id).get().isExists(); | ||
| } | ||
|
|
||
| private IndexRequest indexEvent(JsonElement e) { | ||
| JsonObject obj = e.getAsJsonObject(); | ||
| String type = obj.get("type").getAsString(); | ||
| String id = obj.get("id").getAsString(); | ||
|
|
||
| if (isEventIndexed(id)) { | ||
| return null; | ||
| } | ||
|
|
||
| IndexRequest req = new IndexRequest(index) | ||
| .type(type) | ||
| .id(id).create(false) // we want to overwrite old items | ||
|
|
@@ -237,70 +295,131 @@ private String nextPageURL(URLConnection response) { | |
| return headerData.get("url"); | ||
| } | ||
|
|
||
| private void addAuthHeader(URLConnection connection) { | ||
| if (username == null || password == null) { | ||
| return; | ||
| } | ||
| String auth = String.format("%s:%s", username, password); | ||
| String encoded = Base64.encodeBytes(auth.getBytes()); | ||
| connection.setRequestProperty("Authorization", "Basic " + encoded); | ||
| } | ||
|
|
||
| private boolean getData(String fmt, String type) { | ||
| return getData(fmt, type, null); | ||
| } | ||
|
|
||
| private boolean getData(String fmt, String type, String since) { | ||
| try { | ||
| URL url; | ||
| if (since != null) { | ||
| url = new URL(String.format(fmt, owner, repository, since)); | ||
| } else { | ||
| url = new URL(String.format(fmt, owner, repository)); | ||
| } | ||
| HttpURLConnection connection = (HttpURLConnection) url.openConnection(); | ||
| addAuthHeader(connection); | ||
| if (type.equals("event")) { | ||
|
There was a problem hiding this comment. Indeed, it is a bit unintuitive to have this here. :) There was a problem hiding this comment. Yeah, that was exactly what I meant with "stretching the design" :D |
||
| boolean modified = checkAndUpdateETag(connection); | ||
| if (!modified) { | ||
| return false; | ||
| } | ||
| } | ||
| boolean continueIndexing = indexResponse(connection, type); | ||
|
|
||
| while (continueIndexing && morePagesAvailable(connection)) { | ||
| url = new URL(nextPageURL(connection)); | ||
| connection = (HttpURLConnection) url.openConnection(); | ||
| addAuthHeader(connection); | ||
| continueIndexing = indexResponse(connection, type); | ||
| } | ||
| } catch (Exception e) { | ||
| logger.error("Exception in getData", e); | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| private void deleteByType(String type) { | ||
| client.prepareDeleteByQuery(index) | ||
| .setQuery(termQuery("_type", type)) | ||
| .execute() | ||
| .actionGet(); | ||
| } | ||
|
|
||
| /** | ||
| * Gets the creation data of the single newest entry. | ||
| * | ||
| * @return ISO8601 formatted time of most recent entry, or null on empty or error. | ||
| */ | ||
| private String getMostRecentEntry() { | ||
| long totalEntries = client.prepareCount(index).setQuery(matchAllQuery()).execute().actionGet().getCount(); | ||
| if (totalEntries > 0) { | ||
| FilteredQueryBuilder updatedAtQuery = QueryBuilders | ||
| .filteredQuery(QueryBuilders.matchAllQuery(), FilterBuilders.existsFilter("created_at")); | ||
| FieldSortBuilder updatedAtSort = SortBuilders.fieldSort("created_at").order(SortOrder.DESC); | ||
|
|
||
| SearchResponse response = client.prepareSearch(index) | ||
| .setQuery(updatedAtQuery) | ||
| .addSort(updatedAtSort) | ||
| .setSize(1) | ||
| .execute() | ||
| .actionGet(); | ||
|
|
||
| String createdAt = (String) response.getHits().getAt(0).getSource().get("created_at"); | ||
| logger.debug("Most recent event was created at {}", createdAt); | ||
| return createdAt; | ||
| } else { | ||
| // getData will get all data on a null. | ||
| logger.info("No existing entries, assuming first run"); | ||
| return null; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| while (isRunning) { | ||
| // Must be read before getting new events. | ||
| String mostRecentEntry = getMostRecentEntry(); | ||
|
|
||
| logger.debug("Checking for events"); | ||
| if (getData(endpoint + "/repos/%s/%s/events?per_page=1000", | ||
| "event")) { | ||
| logger.debug("First run or new events found, fetching rest of the data"); | ||
| if (mostRecentEntry != null) { | ||
| getData(endpoint + "/repos/%s/%s/issues?state=all&per_page=1000&since=%s", | ||
| "issue", mostRecentEntry); | ||
| } else { | ||
| getData(endpoint + "/repos/%s/%s/issues?state=all&per_page=1000", | ||
| "issue"); | ||
| } | ||
| // delete pull req data - we are only storing open pull reqs | ||
| // and when a pull request is closed we have no way of knowing; | ||
| // this is why we have to delete them and reindex "fresh" ones | ||
| deleteByType("PullRequestData"); | ||
| getData(endpoint + "/repos/%s/%s/pulls", "pullreq"); | ||
|
|
||
| // same for milestones | ||
| deleteByType("MilestoneData"); | ||
| getData(endpoint + "/repos/%s/%s/milestones?per_page=1000", "milestone"); | ||
|
|
||
| // collaborators | ||
| deleteByType("CollaboratorData"); | ||
| getData(endpoint + "/repos/%s/%s/collaborators?per_page=1000", "collaborator"); | ||
|
|
||
| // and for labels - they have IDs based on the MD5 of the contents, so | ||
| // if a property changes, we get a "new" document | ||
| deleteByType("LabelData"); | ||
| getData(endpoint + "/repos/%s/%s/labels?per_page=1000", "label"); | ||
| } else { | ||
| logger.debug("No new events found"); | ||
| } | ||
| try { | ||
| int waitTime = Math.max(pollInterval, userRequestedInterval) * 1000; | ||
| logger.debug("Waiting {} ms before polling for new events", waitTime); | ||
| Thread.sleep(waitTime); // needs milliseconds | ||
| } catch (InterruptedException e) { | ||
| logger.info("Wait interrupted, river was probably stopped"); | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please add a comment saying that this is now optional (see how it is for auth, below).
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Actually, if the user sets a poll interval that is really large (rarely polling), there will be the possibility of losing events, correct?
So maybe it's better to remove that argument altogether? What do you think?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It has always been optional, AFAIK (the default was 3600 before).
I think we should let it stay, especially until we ETag everything. Without it, an unauthenticated user would be 100% sure to hit the request limit, and they might not care about the events at all, i.e. we don't for our project (we use it to do graphs for issue open time). But we should probably warn that it is a very real possibility.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Whoops, my bad, I forgot :). Ok, good point.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've added a bit about it in the readme.