Java > Open Source Codes > com > sun > slamd > example > LDIF2DBImportRateJobClass


1 /*
2  * Sun Public License
3  *
4  * The contents of this file are subject to the Sun Public License Version
5  * 1.0 (the "License"). You may not use this file except in compliance with
6  * the License. A copy of the License is available at http://www.sun.com/
7  *
8  * The Original Code is the SLAMD Distributed Load Generation Engine.
9  * The Initial Developer of the Original Code is Neil A. Wilson.
10  * Portions created by Neil A. Wilson are Copyright (C) 2004.
11  * Some preexisting portions Copyright (C) 2002-2004 Sun Microsystems, Inc.
12  * All Rights Reserved.
13  *
14  * Contributor(s): Neil A. Wilson
15  */
16 package com.sun.slamd.example;
17
18
19
20 import java.io.*;
21 import java.text.*;
22 import java.util.*;
23 import com.sun.slamd.client.*;
24 import com.sun.slamd.job.*;
25 import com.sun.slamd.parameter.*;
26 import com.sun.slamd.stat.*;
27
28
29
30 /**
31  * This class defines a SLAMD job that has the ability to perform an ldif2db
32  * import for the Sun ONE Directory Server and extract statistics from the
33  * progess of the import, including the total number of entries processed,
34  * the average and recent import rates, and the hit ratio.
35  *
36  *
37  * @author Neil A. Wilson
38  */
39 public class LDIF2DBImportRateJobClass
40        extends JobClass
41 {
42   /**
43    * The display name of the stat tracker used to keep track of the average
44    * import rate.
45    */
46   public static final String STAT_TRACKER_AVERAGE_RATE =
47        "Average Import Rate (Entries/Second)";
48
49
50
51   /**
52    * The display name of the stat tracker used to keep track of the recent
53    * import rate.
54    */
55   public static final String STAT_TRACKER_RECENT_RATE =
56        "Recent Import Rate (Entries/Second)";
57
58
59
60   /**
61    * The display name of the stat tracker used to keep track of the number of
62    * entries processed.
63    */
64   public static final String STAT_TRACKER_ENTRIES_PROCESSED =
65        "Total Entries Processed";
66
67
68
69   /**
70    * The display name of the stat tracker used to keep track of the hit ratio.
71    */
72   public static final String STAT_TRACKER_HIT_RATIO = "Hit Ratio (Percent)";
73
74
75
76   /**
77    * The size to use for the read buffer.
78    */
79   public static final int READ_BUFFER_SIZE = 4096;
80
81
82
83   // The parameter that indicates whether to log command output.
84 BooleanParameter logOutputParameter =
85        new BooleanParameter("log_output", "Log Command Output",
86                             "Indicates whether the ldif2db output should be " +
87                             "logged.", true);
88
89   // The parameter that specifies the name of the database into which the data
90 // is to be imported.
91 StringParameter dbNameParameter =
92        new StringParameter("db_name", "Database Name",
93                            "The name of the database into which the data is " +
94                            "to be imported.", true, "userRoot");
95
96   // The parameter that specifies the path to the ldif2db script.
97 StringParameter ldif2dbCommandParmeter =
98        new StringParameter("ldif2db_command", "ldif2db Command",
99                            "The path to the ldif2db command to be executed.",
100                            true, "");
101
102   // The parameter that specifies the path to the LDIF file to import.
103 StringParameter ldifFileParameter =
104        new StringParameter("ldif_file", "LDIF File",
105                            "The path to the LDIF file to be imported.", true,
106                            "");
107
108   // Indicates whether the output of the command should be captured and logged.
109 static boolean logOutput;
110
111   // The placeholder parameter.
112 PlaceholderParameter placeholder = new PlaceholderParameter();
113
114   // The name of the database into which to import the LDIF file.
115 static String dbName;
116
117   // The command to be executed.
118 static String ldif2dbCommand;
119
120   // The LDIF file containing the data to import.
121 static String ldifFile;
122
123   // The buffer used to hold data read from the process output.
124 byte[] readBuffer;
125
126
127   // The stat trackers maintained by this job.
128 PeriodicEventTracker entriesProcessed;
129   PeriodicEventTracker averageRate;
130   PeriodicEventTracker recentRate;
131   PeriodicEventTracker hitRatio;
132
133
134   // The date format object that will be used to parse the date as it is written
135 // to the log file.
136 SimpleDateFormat dateFormat;
137
138
139
140
141   /**
142    * The default constructor used to create a new instance of the job class.
143    * The only thing it should do is to invoke the superclass constructor. All
144    * other initialization should be performed in the <CODE>initialize</CODE>
145    * method.
146    */
147   public LDIF2DBImportRateJobClass()
148   {
149     super();
150   }
151
152
153
154   /**
155    * Retrieves the name of the job performed by this job thread.
156    *
157    * @return The name of the job performed by this job thread.
158    */
159   public String getJobName()
160   {
161     return "ldif2db Import Rate";
162   }
163
164
165
166   /**
167    * Retrieves a description of the job performed by this job thread.
168    *
169    * @return A description of the job performed by this job thread.
170    */
171   public String getJobDescription()
172   {
173     return "This job can be used to import data into a Sun ONE Directory " +
174            "Server using the ldif2db utility.";
175   }
176
177
178
179   /**
180    * Retrieves the name of the category in which this job class exists. This is
181    * used to help arrange the job classes in the administrative interface.
182    *
183    * @return The name of the category in which this job class exists.
184    */
185   public String getJobCategoryName()
186   {
187     return "LDAP";
188   }
189
190
191
192   /**
193    * Provides a means for job classes to have a level of control over the
194    * number of clients that will be used to run a job. If a job class
195    * implements this method and returns a value greater than 0, then that number
196    * of clients will always be used to run the job, and the form allowing the
197    * user to schedule a job will not show the "Number of Clients" field. By
198    * default, the user will be allowed to choose the number of clients.
199    *
200    * @return The number of clients that should be used to run this job, or -1
201    * if the user should be allowed to specify the number of clients.
202    */
203   public int overrideNumClients()
204   {
205     return 1;
206   }
207
208
209
210   /**
211    * Provides a means for job classes to have a level of control over the
212    * number of threads per client that will be used to run a job. If a job
213    * class implements this method and returns a value greater than 0, then that
214    * number of threads per client will always be used to run the job, and the
215    * form allowing the user to schedule a job will not show the "Threads per
216    * Client" field. By default, the user will be allowed to choose the number
217    * of threads per client.
218    *
219    * @return The number of threads per client that should be used to run this
220    * job, or -1 if the user should be allowed to specify the number
221    * of threads per client.
222    */
223   public int overrideThreadsPerClient()
224   {
225     return 1;
226   }
227
228
229
230   /**
231    * Retrieve a parameter list that can be used to determine all of the
232    * customizeable options that are available for this job.
233    *
234    * @return A parameter list that can be used to determine all of the
235    * customizeable options that are available for this job.
236    */
237   public ParameterList getParameterStubs()
238   {
239     Parameter[] parameters = new Parameter[]
240     {
241       placeholder,
242       ldif2dbCommandParmeter,
243       dbNameParameter,
244       ldifFileParameter,
245       logOutputParameter
246     };
247
248     return new ParameterList(parameters);
249   }
250
251
252
253   /**
254    * Retrieves the set of stat trackers that will be maintained by this job
255    * class. The stat trackers returned by this method do not have to actually
256    * contain any statistics -- the display name and stat tracker class should
257    * be the only information that callers of this method should rely upon. Note
258    * that this list can be different from the list of statistics actually
259    * collected by the job in some cases (e.g., if the job may not return all the
260    * stat trackers it advertises in all cases, or if the job may return stat
261    * trackers that it did not advertise), but it is a possibility that only the
262    * stat trackers returned by this method will be accessible for some features
263    * in the SLAMD server.
264    *
265    * @param clientID The client ID that should be used for the
266    * returned stat trackers.
267    * @param threadID The thread ID that should be used for the
268    * returned stat trackers.
269    * @param collectionInterval The collection interval that should be used for
270    * the returned stat trackers.
271    *
272    * @return The set of stat trackers that will be maintained by this job
273    * class.
274    */
275   public StatTracker[] getStatTrackerStubs(String clientID, String threadID,
276                                            int collectionInterval)
277   {
278     return new StatTracker[]
279     {
280       new PeriodicEventTracker(clientID, threadID, STAT_TRACKER_AVERAGE_RATE,
281                                collectionInterval),
282       new PeriodicEventTracker(clientID, threadID, STAT_TRACKER_RECENT_RATE,
283                                collectionInterval),
284       new PeriodicEventTracker(clientID, threadID,
285                                STAT_TRACKER_ENTRIES_PROCESSED,
286                                collectionInterval),
287       new PeriodicEventTracker(clientID, threadID, STAT_TRACKER_HIT_RATIO,
288                                collectionInterval),
289     };
290   }
291
292
293
294   /**
295    * Retrieves the stat trackers that are maintained for this job thread.
296    *
297    * @return The stat trackers that are maintained for this job thread.
298    */
299   public StatTracker[] getStatTrackers()
300   {
301     return new StatTracker[]
302     {
303       averageRate,
304       recentRate,
305       entriesProcessed,
306       hitRatio
307     };
308   }
309
310
311
312   /**
313    * Initializes all of the instance variables that correspond to job
314    * parameters.
315    *
316    * @param clientID The client ID for the current client.
317    * @param parameters The set of parameters that have been defined for this
318    * job.
319    *
320    * @throws UnableToRunException If any part of the initialization fails.
321    */
322   public void initializeClient(String clientID, ParameterList parameters)
323          throws UnableToRunException
324   {
325     ldif2dbCommand = null;
326     ldif2dbCommandParmeter =
327          parameters.getStringParameter(ldif2dbCommandParmeter.getName());
328     if (ldif2dbCommandParmeter != null)
329     {
330       ldif2dbCommand = ldif2dbCommandParmeter.getStringValue();
331     }
332
333
334     ldifFile = null;
335     ldifFileParameter =
336          parameters.getStringParameter(ldifFileParameter.getName());
337     if (ldifFileParameter != null)
338     {
339       ldifFile = ldifFileParameter.getStringValue();
340     }
341
342
343     dbName = null;
344     dbNameParameter = parameters.getStringParameter(dbNameParameter.getName());
345     if (dbNameParameter != null)
346     {
347       dbName = dbNameParameter.getStringValue();
348     }
349
350
351     logOutput = true;
352     logOutputParameter =
353          parameters.getBooleanParameter(logOutputParameter.getName());
354     if (logOutputParameter != null)
355     {
356       logOutput = logOutputParameter.getBooleanValue();
357     }
358   }
359
360
361
362   /**
363    * Initializes this job thread to be used to actually run the job on the
364    * client. The provided parameter list should be processed to customize the
365    * behavior of this job thread, and any other initialization that needs to be
366    * done in order for the job to run should be performed here as well.
367    *
368    * @param clientID The client ID for this job thread.
369    * @param threadID The thread ID for this job thread.
370    * @param collectionInterval The length of time in seconds to use as the
371    * statistics collection interval.
372    * @param parameters The set of parameters provided to this job that
373    * can be used to customize its behavior.
374    *
375    * @throws UnableToRunException If a problem occurs that prevents the thread
376    * from being able to run properly.
377    */
378   public void initializeThread(String clientID, String threadID,
379                                int collectionInterval, ParameterList parameters)
380          throws UnableToRunException
381   {
382     // Create the stat trackers.
383 averageRate = new PeriodicEventTracker(clientID, threadID,
384                                                 STAT_TRACKER_AVERAGE_RATE,
385                                                 collectionInterval);
386     recentRate = new PeriodicEventTracker(clientID, threadID,
387                                                 STAT_TRACKER_RECENT_RATE,
388                                                 collectionInterval);
389     entriesProcessed = new PeriodicEventTracker(clientID, threadID,
390                                                 STAT_TRACKER_ENTRIES_PROCESSED,
391                                                 collectionInterval);
392     hitRatio = new PeriodicEventTracker(clientID, threadID,
393                                                 STAT_TRACKER_HIT_RATIO,
394                                                 collectionInterval);
395
396     averageRate.setFlatBetweenPoints(false);
397     recentRate.setFlatBetweenPoints(false);
398     entriesProcessed.setFlatBetweenPoints(false);
399     hitRatio.setFlatBetweenPoints(false);
400
401
402     // Initialize the read buffer.
403 readBuffer = new byte[READ_BUFFER_SIZE];
404
405
406     // Initialize the date format.
407 dateFormat = new SimpleDateFormat("[dd/MMM/yyyy:HH:mm:ss Z]");
408   }
409
410
411
412   /**
413    * Perform the work of this job thread by executing the specified command.
414    */
415   public void runJob()
416   {
417     Runtime runtime = Runtime.getRuntime();
418     Process process = null;
419
420     try
421     {
422       String [] commandArray = { ldif2dbCommand, "-n", dbName, "-i", ldifFile };
423       process = runtime.exec(commandArray);
424     }
425     catch (IOException ioe)
426     {
427       logMessage("Unable to execute ldif2db command \"" + ldif2dbCommand +
428                  " -n " + dbName + " -i " + ldifFile + "\": " + ioe);
429       indicateStoppedDueToError();
430       return;
431     }
432
433
434     BufferedInputStream stdOutStream =
435          new BufferedInputStream(process.getInputStream());
436     BufferedInputStream stdErrStream =
437          new BufferedInputStream(process.getErrorStream());
438
439     averageRate.startTracker();
440     recentRate.startTracker();
441     entriesProcessed.startTracker();
442     hitRatio.startTracker();
443
444
445     while (true)
446     {
447       try
448       {
449         if (logOutput)
450         {
451           if (stdOutStream.available() > 0)
452           {
453             while ((! shouldStop()) && (stdOutStream.available() > 0))
454             {
455               int bytesRead = stdOutStream.read(readBuffer);
456               String [] outputStrs = byteArrayToStrings(readBuffer, bytesRead);
457               for (int i=0; i < outputStrs.length; i++)
458               {
459                 logMessage("STDOUT: " + outputStrs[i]);
460                 processLine(outputStrs[i]);
461               }
462             }
463           }
464
465           if (stdErrStream.available() > 0)
466           {
467             while ((! shouldStop()) && (stdErrStream.available() > 0))
468             {
469               int bytesRead = stdErrStream.read(readBuffer);
470               String [] errorStrs = byteArrayToStrings(readBuffer, bytesRead);
471               for (int i=0; i < errorStrs.length; i++)
472               {
473                 logMessage("STDERR: " + errorStrs[i]);
474                 processLine(errorStrs[i]);
475               }
476             }
477           }
478         }
479
480         if (shouldStop())
481         {
482           try
483           {
484             stdOutStream.close();
485             stdErrStream.close();
486           } catch (Exception e) {}
487
488           process.destroy();
489           logMessage("Terminated process because the client determined it " +
490                      "should stop running.");
491           break;
492         }
493
494         try
495         {
496           int returnCode = process.exitValue();
497           if (returnCode == 0)
498           {
499             logMessage("Command completed successfully (exit code 0)");
500           }
501           else
502           {
503             logMessage("Command completed abnormally (exit code " +
504                        returnCode + ")");
505             indicateCompletedWithErrors();
506           }
507
508           try
509           {
510             stdOutStream.close();
511             stdErrStream.close();
512           } catch (Exception e) {}
513
514           break;
515         } catch (IllegalThreadStateException itse) {}
516
517         try
518         {
519           Thread.sleep(100);
520         } catch (InterruptedException ie) {}
521       }
522       catch (IOException ioe)
523       {
524         // This could mean that the command is done or that some other error
525 // occurred. Ty to get the return code to see if it completed.
526 boolean completedSuccessfully = false;
527         try
528         {
529           int returnCode = process.exitValue();
530           completedSuccessfully = (returnCode == 0);
531           if (completedSuccessfully)
532           {
533             logMessage("Command completed successfully (exit code 0)");
534           }
535           else
536           {
537             logMessage("Command completed abnormally (exit code " + returnCode +
538                        ")");
539             indicateCompletedWithErrors();
540           }
541         }
542         catch (IllegalThreadStateException itse)
543         {
544           logMessage("Attempt to read process output failed: " + ioe);
545           indicateCompletedWithErrors();
546         }
547
548         break;
549       }
550     }
551
552
553     averageRate.stopTracker();
554     recentRate.stopTracker();
555     entriesProcessed.stopTracker();
556     hitRatio.stopTracker();
557   }
558
559
560
561   /**
562    * Converts the provided byte array into an array of strings, with one string
563    * per line.
564    *
565    * @param byteArray The byte array containing the data to convert to an
566    * array of strings.
567    * @param length The number of bytes to actually use in the byte array.
568    *
569    * @return The array of strings containing the data from the provided byte
570    * array.
571    */
572   private static String [] byteArrayToStrings(byte[] byteArray, int length)
573   {
574     ArrayList stringList = new ArrayList();
575
576     String byteStr = new String (byteArray, 0, length);
577     StringTokenizer tokenizer = new StringTokenizer(byteStr, "\r\n");
578     while (tokenizer.hasMoreTokens())
579     {
580       stringList.add(tokenizer.nextToken());
581     }
582
583     String [] returnStrings = new String [stringList.size()];
584     stringList.toArray(returnStrings);
585     return returnStrings;
586   }
587
588
589
590   /**
591    * Processes the provided line of output from the ldif2db command and extracts
592    * information about the progress of the import, if the line contains the
593    * appropriate information.
594    *
595    * @param line The line of output captured from ldif2db.
596    */
597   private void processLine(String line)
598   {
599     try
600     {
601       if (line.indexOf(" -- average rate ") < 0)
602       {
603         return;
604       }
605
606       int closeBracketPos = line.indexOf(']');
607       String dateStr = line.substring(0, closeBracketPos+1);
608       Date logDate = dateFormat.parse(dateStr);
609       long logTime = logDate.getTime();
610
611
612       int entriesStartPos = line.indexOf(": Processed ", closeBracketPos) + 12;
613       int entriesEndPos = line.indexOf(" entries", entriesStartPos);
614       int numEntries = Integer.parseInt(line.substring(entriesStartPos,
615                                                             entriesEndPos));
616
617       int avgStartPos = line.indexOf(" -- average rate ", entriesEndPos) +
618                            17;
619       int avgEndPos = line.indexOf("/sec", avgStartPos);
620       double avgRate = Double.parseDouble(line.substring(avgStartPos,
621                                                              avgEndPos));
622
623       int rctStartPos = line.indexOf(", recent rate ", avgEndPos) + 14;
624       int rctEndPos = line.indexOf("/sec", rctStartPos);
625       double rctRate = Double.parseDouble(line.substring(rctStartPos,
626                                                              rctEndPos));
627
628       int hitStartPos = line.indexOf(", hit ratio ", rctEndPos) + 12;
629       int hitEndPos = line.indexOf("%", hitStartPos);
630       double hitRate = Double.parseDouble(line.substring(hitStartPos,
631                                                              hitEndPos));
632
633       writeVerbose("Line is \"" + line + "\"");
634       writeVerbose("Timestamp is " + logDate);
635       writeVerbose("Average rate is " + avgRate);
636       writeVerbose("Recent rate is " + rctRate);
637       writeVerbose("Entries processed is " + numEntries);
638       writeVerbose("Hit ratio is " + hitRate);
639       averageRate.update(logTime, avgRate);
640       recentRate.update(logTime, rctRate);
641       entriesProcessed.update(logTime, numEntries);
642       hitRatio.update(logTime, hitRate);
643     }
644     catch (Exception e)
645     {
646       writeVerbose("Unable to parse line \"" + line + "\" -- " +
647                    stackTraceToString(e));
648     }
649   }
650 }
651
652

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags