4545import java .util .Arrays ;
4646import java .util .Collections ;
4747import java .util .List ;
48+ import java .util .Map ;
4849import java .util .Set ;
4950import java .util .concurrent .ConcurrentHashMap ;
51+ import java .util .concurrent .TimeUnit ;
5052import java .util .concurrent .atomic .AtomicLong ;
5153import java .util .concurrent .atomic .AtomicReference ;
5254
@@ -60,6 +62,9 @@ public abstract class AbstractHttpServerTransport extends AbstractLifecycleCompo
6062 private static final Logger logger = LogManager .getLogger (AbstractHttpServerTransport .class );
6163 private static final ActionListener <Void > NO_OP = ActionListener .wrap (() -> {});
6264
65+ private static final long PRUNE_THROTTLE_INTERVAL = TimeUnit .SECONDS .toMillis (60 );
66+ private static final long MAX_CLIENT_STATS_AGE = TimeUnit .MINUTES .toMillis (5 );
67+
6368 protected final Settings settings ;
6469 public final HttpHandlingSettings handlingSettings ;
6570 protected final NetworkService networkService ;
@@ -78,10 +83,12 @@ public abstract class AbstractHttpServerTransport extends AbstractLifecycleCompo
7883 private final AtomicLong totalChannelsAccepted = new AtomicLong ();
7984 private final Set <HttpChannel > httpChannels = Collections .newSetFromMap (new ConcurrentHashMap <>());
8085 private final Set <HttpServerChannel > httpServerChannels = Collections .newSetFromMap (new ConcurrentHashMap <>());
86+ private final Map <Integer , HttpStats .ClientStats > httpChannelStats = new ConcurrentHashMap <>();
8187
8288 private final HttpTracer tracer ;
8389
8490 private volatile long slowLogThresholdMs ;
91+ protected volatile long lastClientStatsPruneTime ;
8592
8693 protected AbstractHttpServerTransport (Settings settings , NetworkService networkService , BigArrays bigArrays , ThreadPool threadPool ,
8794 NamedXContentRegistry xContentRegistry , Dispatcher dispatcher , ClusterSettings clusterSettings ) {
@@ -128,7 +135,26 @@ public HttpInfo info() {
128135
129136 @ Override
130137 public HttpStats stats () {
131- return new HttpStats (httpChannels .size (), totalChannelsAccepted .get ());
138+ pruneClientStats (false );
139+ return new HttpStats (new ArrayList <>(httpChannelStats .values ()), httpChannels .size (), totalChannelsAccepted .get ());
140+ }
141+
142+ /**
143+ * Prunes client stats of entries that have been disconnected for more than five minutes.
144+ *
145+ * @param throttled When true, executes the prune process only if more than 60 seconds has elapsed since the last execution.
146+ */
147+ void pruneClientStats (boolean throttled ) {
148+ if (throttled == false || (threadPool .relativeTimeInMillis () - lastClientStatsPruneTime > PRUNE_THROTTLE_INTERVAL )) {
149+ long nowMillis = threadPool .absoluteTimeInMillis ();
150+ for (var statsEntry : httpChannelStats .entrySet ()) {
151+ long closedTimeMillis = statsEntry .getValue ().closedTimeMillis ;
152+ if (closedTimeMillis > 0 && (nowMillis - closedTimeMillis > MAX_CLIENT_STATS_AGE )) {
153+ httpChannelStats .remove (statsEntry .getKey ());
154+ }
155+ }
156+ lastClientStatsPruneTime = threadPool .relativeTimeInMillis ();
157+ }
132158 }
133159
134160 protected void bindServer () {
@@ -291,7 +317,23 @@ protected void serverAcceptedChannel(HttpChannel httpChannel) {
291317 boolean addedOnThisCall = httpChannels .add (httpChannel );
292318 assert addedOnThisCall : "Channel should only be added to http channel set once" ;
293319 totalChannelsAccepted .incrementAndGet ();
294- httpChannel .addCloseListener (ActionListener .wrap (() -> httpChannels .remove (httpChannel )));
320+ httpChannelStats .put (
321+ HttpStats .ClientStats .getChannelKey (httpChannel ),
322+ new HttpStats .ClientStats (threadPool .absoluteTimeInMillis ())
323+ );
324+ httpChannel .addCloseListener (ActionListener .wrap (() -> {
325+ try {
326+ httpChannels .remove (httpChannel );
327+ HttpStats .ClientStats clientStats = httpChannelStats .get (HttpStats .ClientStats .getChannelKey (httpChannel ));
328+ if (clientStats != null ) {
329+ clientStats .closedTimeMillis = threadPool .absoluteTimeInMillis ();
330+ }
331+ } catch (Exception e ) {
332+ // the listener code about should never throw
333+ logger .trace ("error removing HTTP channel listener" , e );
334+ }
335+ }));
336+ pruneClientStats (true );
295337 logger .trace (() -> new ParameterizedMessage ("Http channel accepted: {}" , httpChannel ));
296338 }
297339
@@ -302,6 +344,7 @@ protected void serverAcceptedChannel(HttpChannel httpChannel) {
302344 * @param httpChannel that received the http request
303345 */
304346 public void incomingRequest (final HttpRequest httpRequest , final HttpChannel httpChannel ) {
347+ updateClientStats (httpRequest , httpChannel );
305348 final long startTime = threadPool .relativeTimeInMillis ();
306349 try {
307350 handleIncomingRequest (httpRequest , httpChannel , httpRequest .getInboundException ());
@@ -315,6 +358,41 @@ public void incomingRequest(final HttpRequest httpRequest, final HttpChannel htt
315358 }
316359 }
317360
361+ void updateClientStats (final HttpRequest httpRequest , final HttpChannel httpChannel ) {
362+ HttpStats .ClientStats clientStats = httpChannelStats .get (HttpStats .ClientStats .getChannelKey (httpChannel ));
363+ if (clientStats != null ) {
364+ if (clientStats .agent == null ) {
365+ if (hasAtLeastOneHeaderValue (httpRequest , "x-elastic-product-origin" )) {
366+ clientStats .agent = httpRequest .getHeaders ().get ("x-elastic-product-origin" ).get (0 );
367+ } else if (hasAtLeastOneHeaderValue (httpRequest , "User-Agent" )) {
368+ clientStats .agent = httpRequest .getHeaders ().get ("User-Agent" ).get (0 );
369+ }
370+ }
371+ if (clientStats .localAddress == null ) {
372+ clientStats .localAddress = NetworkAddress .format (httpChannel .getLocalAddress ());
373+ clientStats .remoteAddress = NetworkAddress .format (httpChannel .getRemoteAddress ());
374+ }
375+ if (clientStats .forwardedFor == null ) {
376+ if (hasAtLeastOneHeaderValue (httpRequest , "x-forwarded-for" )) {
377+ clientStats .forwardedFor = httpRequest .getHeaders ().get ("x-forwarded-for" ).get (0 );
378+ }
379+ }
380+ if (clientStats .opaqueId == null ) {
381+ if (hasAtLeastOneHeaderValue (httpRequest , "x-opaque-id" )) {
382+ clientStats .opaqueId = httpRequest .getHeaders ().get ("x-opaque-id" ).get (0 );
383+ }
384+ }
385+ clientStats .lastRequestTimeMillis = threadPool .absoluteTimeInMillis ();
386+ clientStats .lastUri = httpRequest .uri ();
387+ clientStats .requestCount .increment ();
388+ clientStats .requestSizeBytes .add (httpRequest .content ().length ());
389+ }
390+ }
391+
392+ private static boolean hasAtLeastOneHeaderValue (final HttpRequest request , final String header ) {
393+ return request .getHeaders ().containsKey (header ) && request .getHeaders ().get (header ).size () > 0 ;
394+ }
395+
318396 // Visible for testing
319397 void dispatchRequest (final RestRequest restRequest , final RestChannel channel , final Throwable badRequestCause ) {
320398 final ThreadContext threadContext = threadPool .getThreadContext ();
0 commit comments