1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
package com.p4square.groupsindexer;
import com.amazonaws.auth.AWS4Signer;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
import com.amazonaws.http.AWSRequestSigningApacheInterceptor;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.amazonaws.services.lambda.runtime.events.ScheduledEvent;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.p4square.ccbapi.CCBAPI;
import com.p4square.ccbapi.CCBAPIClient;
import com.p4square.ccbapi.model.*;
import com.p4square.groupsindexer.model.GroupSearchDocument;
import com.p4square.groupsindexer.model.GroupSearchDocumentAdapter;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequestInterceptor;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URL;
/**
* UpdateIndexes is a scheduled lambda which populates the groups search index.
*
* Required (custom) environment variables:
* <ul>
* <li>CCBAPIURL</li>
* <li>CCBAPIUser</li>
* <li>CCBAPIPassword</li>
* <li>ES_URL</li>
* <li>IMAGE_BUCKET</li>
* </ul>
*
*/
public class UpdateIndexes implements RequestHandler<ScheduledEvent, String> {
private static final AWSCredentialsProvider credentialsProvider = new DefaultAWSCredentialsProviderChain();
private static final Logger LOG = LogManager.getLogger(UpdateIndexes.class);
private static final GroupSearchDocumentAdapter ADAPTER = new GroupSearchDocumentAdapter();
private static final ObjectMapper MAPPER = new ObjectMapper();
private final String imageBucket;
private final CCBAPI ccbClient;
private final RestHighLevelClient esClient;
private final AmazonS3 s3Client;
public UpdateIndexes() throws Exception {
// Setup CCB Client
final String CCBAPIURL = System.getenv("CCBAPIURL");
final String CCBAPIUser = System.getenv("CCBAPIUser");
final String CCBAPIPassword = System.getenv("CCBAPIPassword");
ccbClient = new CCBAPIClient(new URI(CCBAPIURL), CCBAPIUser, CCBAPIPassword);
// Setup ElasticSearch client
final String ES_URL = System.getenv("ES_URL");
AWS4Signer signer = new AWS4Signer();
signer.setServiceName("es");
signer.setRegionName(System.getenv("AWS_DEFAULT_REGION"));
HttpRequestInterceptor interceptor = new AWSRequestSigningApacheInterceptor(signer.getServiceName(), signer, credentialsProvider);
esClient = new RestHighLevelClient(RestClient
.builder(HttpHost.create(ES_URL))
.setHttpClientConfigCallback(hacb -> hacb.addInterceptorLast(interceptor)));
// Setup S3 Client
imageBucket = System.getenv("IMAGE_BUCKET");
s3Client = AmazonS3ClientBuilder.defaultClient();
}
@Override
public String handleRequest(ScheduledEvent s3Event, Context context) {
try {
GetGroupProfilesResponse response = ccbClient.getGroupProfiles(
new GetGroupProfilesRequest()
.withIncludeImageUrl(true)
.withIncludeParticipants(false));
final BulkRequest indexRequest = new BulkRequest();
for (GroupProfile profile : response.getGroups()) {
if (!profile.isActive() ||
!profile.isPublicSearchListed() ||
profile.getInteractionType() != InteractionType.MEMBERS_INTERACT) {
LOG.info("Skipping inactive/unlisted group " + profile.getName());
continue;
}
// Transform GroupProfile to Search Document.
final GroupSearchDocument document = ADAPTER.apply(profile);
// Save GroupProfile image.
document.setImageUrl(null);
if (profile.getImageUrl() != null && !profile.getImageUrl().isEmpty()) {
final String imageKey = "group-images/group-" + profile.getId();
InputStream in = null;
try {
final URL imageUrl = new URL(profile.getImageUrl());
in = imageUrl.openStream();
s3Client.putObject(imageBucket, imageKey, in, null);
document.setImageUrl(imageKey);
} catch (Exception e) {
LOG.error("Failed to upload image for group " + profile.getId(), e);
} finally {
if (in != null) {
try {
in.close();
} catch (IOException e) {
// Ignore
}
}
}
}
// Add request to batch.
indexRequest.add(Requests
.indexRequest("groups")
.type("group")
.id(String.valueOf(document.getId()))
.source(MAPPER.writeValueAsString(document), XContentType.JSON));
}
BulkResponse esResponse = esClient.bulk(indexRequest);
if (esResponse.hasFailures()) {
throw new RuntimeException(esResponse.buildFailureMessage());
}
LOG.info("Updated search index. Found " + response.getGroups().size() + " groups.");
return "ok";
} catch (IOException e) {
LOG.error("Unexpected Exception: " + e.getMessage(), e);
throw new RuntimeException(e);
}
}
}
|
