1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.chukwa.database;
19
20 import junit.framework.TestCase;
21 import java.util.Calendar;
22 import org.apache.hadoop.chukwa.database.Macro;
23 import org.apache.hadoop.chukwa.util.DatabaseWriter;
24 import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
25 import org.apache.hadoop.chukwa.util.ExceptionUtil;
26 import org.apache.hadoop.chukwa.database.Aggregator;
27 import org.apache.hadoop.chukwa.database.TableCreator;
28 import java.io.BufferedReader;
29 import java.io.File;
30 import java.io.FileReader;
31 import java.io.IOException;
32 import java.sql.ResultSet;
33 import java.sql.SQLException;
34 import java.util.ArrayList;
35
36 public class TestDatabaseDataExpiration extends TestCase {
37
38 long[] timeWindow = {7, 30, 91, 365, 3650};
39 String[] tables = {"system_metrics","disk","cluster_system_metrics","cluster_disk","mr_job","mr_task","dfs_namenode","dfs_datanode","dfs_fsnamesystem","dfs_throughput","hadoop_jvm","hadoop_mapred","hdfs_usage"};
40 String cluster = "demo";
41 long current = Calendar.getInstance().getTimeInMillis();
42
43 public void setUp() throws Exception {
44 System.setProperty("CLUSTER","demo");
45 DatabaseWriter db = new DatabaseWriter(cluster);
46 String buffer = "";
47 File aFile = new File(System.getenv("CHUKWA_CONF_DIR")
48 + File.separator + "database_create_tables.sql");
49 buffer = readFile(aFile);
50 String tables[] = buffer.split(";");
51 for(String table : tables) {
52 if(table.length()>5) {
53 db.execute(table);
54 }
55 }
56 db.close();
57 for(int i=0;i<timeWindow.length;i++) {
58 TableCreator tc = new TableCreator();
59 long start = current;
60 long end = current + (timeWindow[i]*1440*60*1000);
61 tc.createTables(start, end);
62 }
63 }
64
65 public void tearDown() {
66 DatabaseWriter db = null;
67 try {
68 db = new DatabaseWriter(cluster);
69 ResultSet rs = db.query("show tables");
70 ArrayList<String> list = new ArrayList<String>();
71 while(rs.next()) {
72 String table = rs.getString(1);
73 list.add(table);
74 }
75 for(String table : list) {
76 db.execute("drop table "+table);
77 }
78 } catch(Throwable ex) {
79 } finally {
80 if(db!=null) {
81 db.close();
82 }
83 }
84 }
85
86 public void verifyTable(String table) {
87 ChukwaConfiguration cc = new ChukwaConfiguration();
88 String query = "select * from ["+table+"];";
89 Macro mp = new Macro(current,query);
90 query = mp.toString();
91 try {
92 DatabaseWriter db = new DatabaseWriter(cluster);
93 ResultSet rs = db.query(query);
94 while(rs.next()) {
95 int i = 1;
96 String value = rs.getString(i);
97 }
98 db.close();
99 } catch(SQLException ex) {
100 fail("SQL Exception: "+ExceptionUtil.getStackTrace(ex));
101 }
102 }
103
104 public String readFile(File aFile) {
105 StringBuffer contents = new StringBuffer();
106 try {
107 BufferedReader input = new BufferedReader(new FileReader(aFile));
108 try {
109 String line = null;
110 while ((line = input.readLine()) != null) {
111 contents.append(line);
112 contents.append(System.getProperty("line.separator"));
113 }
114 } finally {
115 input.close();
116 }
117 } catch (IOException ex) {
118 ex.printStackTrace();
119 }
120 return contents.toString();
121 }
122
123 public void testDataExpiration() {
124 for(int i=0;i<timeWindow.length;i++) {
125 long start = current + (365*1440*60*1000);
126 long end = start + (timeWindow[i]*1440*60*1000);
127 try {
128 DataExpiration de = new DataExpiration();
129 de.dropTables(start, end);
130 } catch(Throwable ex) {
131 fail("SQL Exception: "+ExceptionUtil.getStackTrace(ex));
132 }
133 assertTrue("DataExpiration executed.", true);
134 DatabaseWriter db = null;
135 try {
136 db = new DatabaseWriter(cluster);
137 String query = "select * from [system_metrics];";
138 Macro mp = new Macro(current,query);
139 query = mp.toString();
140 ResultSet rs = db.query(query);
141 } catch(SQLException ex) {
142 assertTrue("Table is not suppose to exist.",true);
143 db.close();
144 }
145 }
146 }
147
148 }