View Javadoc

    1   /*
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  You may obtain a copy of the License at
    8    *
    9    *      http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   */
    17
    18  package org.apache.log4j.net;
    19
    20  import java.io.IOException;
    21  import java.io.InterruptedIOException;
    22  import java.io.ObjectOutputStream;
    23  import java.net.InetAddress;
    24  import java.net.ServerSocket;
    25  import java.net.Socket;
    26  import java.net.SocketException;
    27  import java.util.Vector;
    28
    29  import org.apache.log4j.AppenderSkeleton;
    30  import org.apache.log4j.helpers.CyclicBuffer;
    31  import org.apache.log4j.helpers.LogLog;
    32  import org.apache.log4j.spi.LoggingEvent;
    33
    34  /**
    35    Sends {@link LoggingEvent} objects to a set of remote log servers,
    36    usually a {@link SocketNode SocketNodes}.
    37
    38    <p>Acts just like {@link SocketAppender} except that instead of
    39    connecting to a given remote log server,
    40    <code>SocketHubAppender</code> accepts connections from the remote
    41    log servers as clients.  It can accept more than one connection.
    42    When a log event is received, the event is sent to the set of
    43    currently connected remote log servers. Implemented this way it does
    44    not require any update to the configuration file to send data to
    45    another remote log server. The remote log server simply connects to
    46    the host and port the <code>SocketHubAppender</code> is running on.
    47
    48    <p>The <code>SocketHubAppender</code> does not store events such
    49    that the remote side will events that arrived after the
    50    establishment of its connection. Once connected, events arrive in
    51    order as guaranteed by the TCP protocol.
    52
    53    <p>This implementation borrows heavily from the {@link
    54    SocketAppender}.
    55
    56    <p>The SocketHubAppender has the following characteristics:
    57
    58    <ul>
    59
    60    <p><li>If sent to a {@link SocketNode}, logging is non-intrusive as
    61    far as the log event is concerned. In other words, the event will be
    62    logged with the same time stamp, {@link org.apache.log4j.NDC},
    63    location info as if it were logged locally.
    64
    65    <p><li><code>SocketHubAppender</code> does not use a layout. It
    66    ships a serialized {@link LoggingEvent} object to the remote side.
    67
    68    <p><li><code>SocketHubAppender</code> relies on the TCP
    69    protocol. Consequently, if the remote side is reachable, then log
    70    events will eventually arrive at remote client.
    71
    72    <p><li>If no remote clients are attached, the logging requests are
    73    simply dropped.
    74
    75    <p><li>Logging events are automatically <em>buffered</em> by the
    76    native TCP implementation. This means that if the link to remote
    77    client is slow but still faster than the rate of (log) event
    78    production, the application will not be affected by the slow network
    79    connection. However, if the network connection is slower then the
    80    rate of event production, then the local application can only
    81    progress at the network rate. In particular, if the network link to
    82    the the remote client is down, the application will be blocked.
    83
    84    <p>On the other hand, if the network link is up, but the remote
    85    client is down, the client will not be blocked when making log
    86    requests but the log events will be lost due to client
    87    unavailability.
    88
    89    <p>The single remote client case extends to multiple clients
    90    connections. The rate of logging will be determined by the slowest
    91    link.
    92
    93    <p><li>If the JVM hosting the <code>SocketHubAppender</code> exits
    94    before the <code>SocketHubAppender</code> is closed either
    95    explicitly or subsequent to garbage collection, then there might
    96    be untransmitted data in the pipe which might be lost. This is a
    97    common problem on Windows based systems.
    98
    99    <p>To avoid lost data, it is usually sufficient to {@link #close}
    100   the <code>SocketHubAppender</code> either explicitly or by calling
    101   the {@link org.apache.log4j.LogManager#shutdown} method before
    102   exiting the application.
    103
    104   </ul>
    105
    106   @author Mark Womack */
    107
    108 public class SocketHubAppender extends AppenderSkeleton {
    109
    110   /**
    111      The default port number of the ServerSocket will be created on. */
    112   static final int DEFAULT_PORT = 4560;
    113
    114   private int port = DEFAULT_PORT;
    115   private Vector oosList = new Vector();
    116   private ServerMonitor serverMonitor = null;
    117   private boolean locationInfo = false;
    118   private CyclicBuffer buffer = null;
    119   private String application;
    120   private boolean advertiseViaMulticastDNS;
    121   private ZeroConfSupport zeroConf;
    122
    123   /**
    124    * The MulticastDNS zone advertised by a SocketHubAppender
    125    */
    126   public static final String ZONE = "_log4j_obj_tcpaccept_appender.local.";
    127
    128
    129   public SocketHubAppender() { }
    130
    131   /**
    132      Connects to remote server at <code>address</code> and <code>port</code>. */
    133   public
    134   SocketHubAppender(int _port) {
    135     port = _port;
    136     startServer();
    137   }
    138
    139   /**
    140      Set up the socket server on the specified port.  */
    141   public
    142   void activateOptions() {
    143     if (advertiseViaMulticastDNS) {
    144       zeroConf = new ZeroConfSupport(ZONE, port, getName());
    145       zeroConf.advertise();
    146     }
    147     startServer();
    148   }
    149
    150   /**
    151      Close this appender.
    152      <p>This will mark the appender as closed and
    153      call then {@link #cleanUp} method. */
    154   synchronized
    155   public
    156   void close() {
    157     if(closed)
    158       return;
    159
    160 	LogLog.debug("closing SocketHubAppender " + getName());
    161     this.closed = true;
    162     if (advertiseViaMulticastDNS) {
    163       zeroConf.unadvertise();
    164     }
    165     cleanUp();
    166
    167 	LogLog.debug("SocketHubAppender " + getName() + " closed");
    168   }
    169
    170   /**
    171      Release the underlying ServerMonitor thread, and drop the connections
    172      to all connected remote servers. */
    173   public
    174   void cleanUp() {
    175     // stop the monitor thread
    176 	LogLog.debug("stopping ServerSocket");
    177     serverMonitor.stopMonitor();
    178     serverMonitor = null;
    179
    180     // close all of the connections
    181 	LogLog.debug("closing client connections");
    182     while (oosList.size() != 0) {
    183       ObjectOutputStream oos = (ObjectOutputStream)oosList.elementAt(0);
    184       if(oos != null) {
    185         try {
    186         	oos.close();
    187         } catch(InterruptedIOException e) {
    188             Thread.currentThread().interrupt();
    189             LogLog.error("could not close oos.", e);
    190         } catch(IOException e) {
    191             LogLog.error("could not close oos.", e);
    192         }
    193
    194         oosList.removeElementAt(0);
    195       }
    196     }
    197   }
    198
    199   /**
    200     Append an event to all of current connections. */
    201   public
    202   void append(LoggingEvent event) {
    203     if (event != null) {
    204       // set up location info if requested
    205       if (locationInfo) {
    206         event.getLocationInformation();
    207       }
    208       if (application != null) {
    209           event.setProperty("application", application);
    210         }
    211         event.getNDC();
    212         event.getThreadName();
    213         event.getMDCCopy();
    214         event.getRenderedMessage();
    215         event.getThrowableStrRep();
    216
    217       if (buffer != null) {
    218         buffer.add(event);
    219       }
    220     }
    221
    222     // if no event or no open connections, exit now
    223     if ((event == null) || (oosList.size() == 0)) {
    224       return;
    225     }
    226
    227 	// loop through the current set of open connections, appending the event to each
    228     for (int streamCount = 0; streamCount < oosList.size(); streamCount++) {
    229
    230       ObjectOutputStream oos = null;
    231       try {
    232         oos = (ObjectOutputStream)oosList.elementAt(streamCount);
    233       }
    234       catch (ArrayIndexOutOfBoundsException e) {
    235         // catch this, but just don't assign a value
    236         // this should not really occur as this method is
    237         // the only one that can remove oos's (besides cleanUp).
    238       }
    239
    240       // list size changed unexpectedly? Just exit the append.
    241       if (oos == null)
    242         break;
    243
    244       try {
    245       	oos.writeObject(event);
    246       	oos.flush();
    247     	// Failing to reset the object output stream every now and
    248     	// then creates a serious memory leak.
    249     	// right now we always reset. TODO - set up frequency counter per oos?
    250     	oos.reset();
    251       }
    252       catch(IOException e) {
    253         if (e instanceof InterruptedIOException) {
    254             Thread.currentThread().interrupt();
    255         }
    256           // there was an io exception so just drop the connection
    257       	oosList.removeElementAt(streamCount);
    258       	LogLog.debug("dropped connection");
    259
    260       	// decrement to keep the counter in place (for loop always increments)
    261       	streamCount--;
    262       }
    263     }
    264   }
    265
    266   /**
    267      The SocketHubAppender does not use a layout. Hence, this method returns
    268      <code>false</code>. */
    269   public
    270   boolean requiresLayout() {
    271     return false;
    272   }
    273
    274   /**
    275      The <b>Port</b> option takes a positive integer representing
    276      the port where the server is waiting for connections. */
    277   public
    278   void setPort(int _port) {
    279     port = _port;
    280 	}
    281
    282   /**
    283    * The <b>App</b> option takes a string value which should be the name of the application getting logged. If property was already set (via system
    284    * property), don't set here.
    285    */
    286   public
    287   void setApplication(String lapp) {
    288     this.application = lapp;
    289   }
    290
    291   /**
    292    * Returns value of the <b>Application</b> option.
    293    */
    294   public
    295   String getApplication() {
    296     return application;
    297   }
    298
    299   /**
    300      Returns value of the <b>Port</b> option. */
    301   public
    302   int getPort() {
    303     return port;
    304   }
    305
    306   /**
    307    * The <b>BufferSize</b> option takes a positive integer representing the number of events this appender will buffer and send to newly connected
    308    * clients.
    309    */
    310   public
    311   void setBufferSize(int _bufferSize) {
    312     buffer = new CyclicBuffer(_bufferSize);
    313   }
    314
    315   /**
    316    * Returns value of the <b>bufferSize</b> option.
    317    */
    318   public
    319   int getBufferSize() {
    320     if (buffer == null) {
    321       return 0;
    322     } else {
    323       return buffer.getMaxSize();
    324     }
    325   }
    326
    327   /**
    328      The <b>LocationInfo</b> option takes a boolean value. If true,
    329      the information sent to the remote host will include location
    330      information. By default no location information is sent to the server. */
    331   public
    332   void setLocationInfo(boolean _locationInfo) {
    333     locationInfo = _locationInfo;
    334   }
    335
    336   /**
    337      Returns value of the <b>LocationInfo</b> option. */
    338   public
    339   boolean getLocationInfo() {
    340     return locationInfo;
    341   }
    342
    343   public void setAdvertiseViaMulticastDNS(boolean advertiseViaMulticastDNS) {
    344     this.advertiseViaMulticastDNS = advertiseViaMulticastDNS;
    345   }
    346
    347   public boolean isAdvertiseViaMulticastDNS() {
    348     return advertiseViaMulticastDNS;
    349   }
    350
    351   /**
    352     Start the ServerMonitor thread. */
    353   private
    354   void startServer() {
    355     serverMonitor = new ServerMonitor(port, oosList);
    356   }
    357
    358   /**
    359    * Creates a server socket to accept connections.
    360    * @param socketPort port on which the socket should listen, may be zero.
    361    * @return new socket.
    362    * @throws IOException IO error when opening the socket.
    363    */
    364   protected ServerSocket createServerSocket(final int socketPort) throws IOException {
    365       return new ServerSocket(socketPort);
    366   }
    367
    368   /**
    369     This class is used internally to monitor a ServerSocket
    370     and register new connections in a vector passed in the
    371     constructor. */
    372   private
    373   class ServerMonitor implements Runnable {
    374     private int port;
    375     private Vector oosList;
    376     private boolean keepRunning;
    377     private Thread monitorThread;
    378
    379     /**
    380       Create a thread and start the monitor. */
    381     public
    382     ServerMonitor(int _port, Vector _oosList) {
    383       port = _port;
    384       oosList = _oosList;
    385       keepRunning = true;
    386       monitorThread = new Thread(this);
    387       monitorThread.setDaemon(true);
    388       monitorThread.setName("SocketHubAppender-Monitor-" + port);
    389       monitorThread.start();
    390     }
    391
    392     /**
    393       Stops the monitor. This method will not return until
    394       the thread has finished executing. */
    395     public
    396     synchronized
    397     void stopMonitor() {
    398       if (keepRunning) {
    399     	LogLog.debug("server monitor thread shutting down");
    400         keepRunning = false;
    401         try {
    402           monitorThread.join();
    403         }
    404         catch (InterruptedException e) {
    405             Thread.currentThread().interrupt();
    406           // do nothing?
    407         }
    408
    409         // release the thread
    410         monitorThread = null;
    411     	LogLog.debug("server monitor thread shut down");
    412       }
    413     }
    414
    415     private
    416     void sendCachedEvents(ObjectOutputStream stream) throws IOException {
    417       if (buffer != null) {
    418         for (int i = 0; i < buffer.length(); i++) {
    419           stream.writeObject(buffer.get(i));
    420         }
    421         stream.flush();
    422         stream.reset();
    423       }
    424     }
    425
    426     /**
    427       Method that runs, monitoring the ServerSocket and adding connections as
    428       they connect to the socket. */
    429     public
    430     void run() {
    431       ServerSocket serverSocket = null;
    432       try {
    433         serverSocket = createServerSocket(port);
    434         serverSocket.setSoTimeout(1000);
    435       }
    436       catch (Exception e) {
    437         if (e instanceof InterruptedIOException || e instanceof InterruptedException) {
    438             Thread.currentThread().interrupt();
    439         }
    440         LogLog.error("exception setting timeout, shutting down server socket.", e);
    441         keepRunning = false;
    442         return;
    443       }
    444
    445       try {
    446     	try {
    447         	serverSocket.setSoTimeout(1000);
    448     	}
    449     	catch (SocketException e) {
    450           LogLog.error("exception setting timeout, shutting down server socket.", e);
    451           return;
    452     	}
    453
    454     	while (keepRunning) {
    455           Socket socket = null;
    456           try {
    457             socket = serverSocket.accept();
    458           }
    459           catch (InterruptedIOException e) {
    460             // timeout occurred, so just loop
    461           }
    462           catch (SocketException e) {
    463             LogLog.error("exception accepting socket, shutting down server socket.", e);
    464             keepRunning = false;
    465           }
    466           catch (IOException e) {
    467             LogLog.error("exception accepting socket.", e);
    468           }
    469
    470           // if there was a socket accepted
    471           if (socket != null) {
    472             try {
    473               InetAddress remoteAddress = socket.getInetAddress();
    474               LogLog.debug("accepting connection from " + remoteAddress.getHostName()
    475 			   + " (" + remoteAddress.getHostAddress() + ")");
    476
    477               // create an ObjectOutputStream
    478               ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
    479               if (buffer != null && buffer.length() > 0) {
    480                 sendCachedEvents(oos);
    481               }
    482
    483               // add it to the oosList.  OK since Vector is synchronized.
    484               oosList.addElement(oos);
    485             } catch (IOException e) {
    486               if (e instanceof InterruptedIOException) {
    487                     Thread.currentThread().interrupt();
    488               }
    489               LogLog.error("exception creating output stream on socket.", e);
    490             }
    491           }
    492         }
    493       }
    494       finally {
    495     	// close the socket
    496     	try {
    497     		serverSocket.close();
    498     	} catch(InterruptedIOException e) {
    499             Thread.currentThread().interrupt();
    500         } catch (IOException e) {
    501     		// do nothing with it?
    502     	}
    503       }
    504     }
    505   }
    506 }
    507