-
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
-
151
-
152
-
153
-
154
-
155
-
156
-
157
-
158
-
159
-
160
-
161
-
162
-
163
-
164
-
165
-
166
-
167
-
168
-
169
-
170
-
171
-
172
-
173
-
174
-
175
-
176
-
177
-
178
-
179
-
180
-
181
-
182
-
183
-
184
-
185
-
186
-
187
-
188
-
189
-
190
-
191
-
192
-
193
-
194
-
195
-
196
-
197
-
198
-
199
-
200
-
201
-
202
-
203
-
204
-
205
-
206
-
207
-
208
-
209
-
210
-
211
-
212
-
213
-
214
-
215
-
216
-
217
-
218
-
219
-
220
-
221
-
222
-
223
-
224
-
225
-
226
-
227
-
228
-
229
-
230
-
231
-
232
-
233
-
234
-
235
-
236
-
237
-
238
-
239
-
240
-
241
-
242
-
243
-
244
-
245
-
246
-
247
-
248
-
249
-
250
-
251
-
252
-
253
-
254
-
255
-
256
-
257
-
258
-
259
-
260
-
261
-
262
-
263
-
264
-
265
-
266
-
267
-
268
-
269
-
270
-
271
-
272
-
273
-
274
-
275
-
276
-
277
-
278
-
279
-
280
-
281
-
282
-
283
-
284
-
285
-
286
-
287
-
288
-
289
-
290
-
291
-
292
-
293
-
294
-
295
-
296
-
297
-
298
-
299
-
300
-
301
-
302
-
303
-
304
-
305
-
306
-
307
-
308
-
309
-
310
-
311
-
312
-
313
-
314
-
315
-
316
-
317
-
318
-
319
-
320
-
321
-
322
-
323
-
324
-
325
-
326
-
327
-
328
-
329
-
330
-
331
-
332
-
333
-
334
-
335
-
336
-
337
-
338
-
339
-
340
-
341
-
342
-
343
-
344
-
345
-
346
-
347
-
348
-
349
-
350
-
351
-
352
-
353
-
354
-
355
-
356
-
357
-
358
-
359
-
360
-
361
-
362
-
363
-
364
-
365
-
366
-
367
-
368
-
369
-
370
-
371
-
372
-
373
-
374
-
375
-
376
-
377
-
378
-
379
-
380
-
381
-
382
-
383
-
384
-
385
-
386
-
387
-
388
-
389
-
390
-
391
-
392
-
393
-
394
-
395
-
396
-
397
-
398
-
399
-
400
-
401
-
402
-
403
-
404
-
405
-
406
-
407
-
408
-
409
-
410
-
411
-
412
-
413
-
414
-
415
-
416
-
417
-
418
-
419
-
420
-
421
-
422
-
423
-
424
-
425
-
426
-
427
-
428
-
429
-
430
-
431
-
432
-
433
-
434
-
435
-
436
-
437
-
438
-
439
-
440
-
441
-
442
-
443
-
444
-
445
-
446
-
447
-
448
-
449
-
450
-
451
-
452
-
453
-
454
-
455
-
456
-
457
-
458
-
459
-
460
-
461
-
462
-
463
-
464
-
465
-
466
-
467
-
468
-
469
-
470
-
471
-
472
-
473
-
474
-
475
-
476
-
477
-
478
-
479
-
480
-
481
-
482
-
483
-
484
-
485
-
486
-
487
-
488
-
489
-
490
-
491
-
492
-
493
-
494
-
495
-
496
-
497
-
498
-
499
-
500
-
501
-
502
-
503
-
504
-
505
-
506
-
507
-
508
-
509
-
510
-
511
-
512
-
513
-
514
-
515
-
516
-
517
-
518
-
519
-
520
-
521
-
522
-
523
-
524
-
525
-
526
-
527
-
528
-
529
-
530
-
531
-
532
-
533
-
534
-
535
-
536
-
537
-
538
-
539
-
540
-
541
-
542
-
543
-
544
-
545
-
546
-
547
-
548
-
549
-
550
-
551
-
552
-
553
-
554
-
555
-
556
-
557
-
558
-
559
-
560
-
561
-
562
-
563
-
564
-
565
-
566
-
567
-
568
-
569
-
570
-
571
-
572
-
573
-
574
-
575
-
576
-
577
-
578
-
579
-
580
-
581
-
582
-
583
-
584
-
585
-
586
-
587
-
588
-
589
-
590
-
591
-
592
-
593
-
594
-
595
-
596
-
597
-
598
-
599
-
600
-
601
-
602
-
603
-
604
-
605
-
606
-
607
-
608
-
609
-
610
-
611
-
612
-
613
-
614
-
615
-
616
-
617
-
618
-
619
-
620
-
621
-
622
-
623
-
624
-
625
-
626
-
627
-
628
-
629
-
630
-
631
-
632
-
633
-
634
-
635
-
636
-
637
-
638
-
639
-
640
-
641
-
642
-
643
-
644
-
645
-
646
-
647
-
648
-
649
-
650
-
651
-
652
-
653
-
654
-
655
-
656
-
657
-
658
-
659
-
660
-
661
-
662
-
663
-
664
-
665
-
666
-
667
-
668
-
669
-
670
-
671
-
672
-
673
-
674
-
675
-
676
-
677
-
678
-
679
-
680
-
681
-
682
-
683
-
684
-
685
-
686
-
687
-
688
-
689
-
690
-
691
-
692
-
693
-
694
-
695
-
696
-
697
-
698
-
699
-
700
-
701
-
702
-
703
-
704
-
705
-
706
-
707
-
708
-
709
-
710
-
711
-
712
-
713
-
714
-
715
-
716
-
717
-
718
-
719
-
720
-
721
-
722
-
723
-
724
-
725
-
726
-
727
-
728
-
729
-
730
-
731
-
732
-
733
-
734
-
735
-
736
-
737
-
738
-
739
-
740
-
741
-
742
-
743
-
744
-
745
-
746
-
747
-
748
-
749
-
750
-
751
-
752
-
753
-
754
-
755
-
756
-
757
-
758
-
759
-
760
-
761
-
762
-
763
-
764
-
765
-
766
-
767
-
768
-
769
-
770
-
771
-
772
-
773
-
774
-
775
-
776
-
777
-
778
-
779
-
780
-
781
-
782
-
783
-
784
-
785
-
786
-
787
-
788
-
789
-
790
-
791
-
792
-
793
-
794
-
795
-
796
-
797
-
798
-
799
-
800
-
801
-
802
-
803
-
804
-
805
-
806
-
807
-
808
-
809
-
810
-
811
-
812
-
813
-
814
-
815
-
816
-
817
-
818
-
819
-
820
-
821
-
822
-
823
-
824
-
825
-
826
-
827
-
828
-
829
-
830
-
831
-
832
-
833
-
834
-
835
-
836
-
837
-
838
-
839
-
840
-
841
-
842
-
843
-
844
-
845
-
846
-
847
-
848
-
849
-
850
-
851
-
852
-
853
-
854
-
855
-
856
-
857
-
858
-
859
-
860
-
861
-
862
-
863
-
864
-
865
-
866
-
867
-
868
-
869
-
870
-
871
-
872
-
873
-
874
-
875
-
876
-
877
-
878
-
879
-
880
-
881
-
882
-
883
-
884
-
885
-
886
-
887
-
888
-
889
-
890
-
891
-
892
-
893
-
894
-
895
-
896
-
897
-
898
-
899
-
900
-
901
-
902
-
903
-
904
-
905
-
906
-
907
-
908
-
909
-
910
-
911
-
912
-
913
-
914
-
915
-
916
-
917
-
918
-
919
-
920
-
921
-
922
-
923
-
924
-
925
-
926
-
927
-
928
-
929
-
930
-
931
-
932
-
933
-
934
-
935
-
936
-
937
-
938
-
939
-
940
-
941
-
942
-
943
-
944
-
945
-
946
-
947
-
948
-
949
-
950
-
951
-
952
-
953
-
954
-
955
-
956
-
957
-
958
-
959
-
960
-
961
-
962
-
963
-
964
-
965
-
966
-
967
-
968
-
969
-
970
-
971
-
972
-
973
-
974
-
975
-
976
-
977
-
978
-
979
-
980
-
981
-
982
-
983
-
984
-
985
-
986
-
987
-
988
-
989
-
990
-
991
-
992
-
993
-
994
-
995
-
996
-
997
-
998
-
999
-
1000
-
1001
-
1002
-
1003
-
1004
-
1005
-
1006
-
1007
-
1008
-
1009
-
1010
-
1011
-
1012
-
1013
-
1014
-
1015
-
1016
-
1017
-
1018
-
1019
-
1020
-
1021
-
1022
-
1023
-
1024
-
1025
-
1026
-
1027
-
1028
-
1029
-
1030
-
1031
-
1032
-
1033
-
1034
-
1035
-
1036
-
1037
-
1038
-
1039
-
1040
-
1041
-
1042
-
1043
-
1044
-
1045
-
1046
-
1047
-
1048
-
1049
-
1050
-
1051
-
1052
-
1053
-
1054
-
1055
-
1056
-
1057
-
1058
-
1059
-
1060
-
1061
-
1062
-
1063
-
1064
-
1065
-
1066
-
1067
-
1068
-
1069
-
1070
-
1071
-
1072
-
1073
-
1074
-
1075
-
1076
-
1077
-
1078
-
1079
-
1080
-
1081
-
1082
-
1083
-
1084
-
1085
-
1086
-
1087
-
1088
-
1089
-
1090
-
1091
-
1092
-
1093
-
1094
-
1095
-
1096
-
1097
-
1098
-
1099
-
1100
-
1101
-
1102
-
1103
-
1104
-
1105
-
1106
-
1107
-
1108
-
1109
-
1110
-
1111
-
1112
-
1113
-
1114
-
1115
-
1116
-
1117
-
1118
-
1119
-
1120
-
1121
-
1122
-
1123
-
1124
-
1125
-
1126
-
1127
-
1128
-
1129
-
1130
-
1131
-
1132
-
1133
-
1134
-
1135
-
1136
-
1137
-
1138
-
1139
-
1140
-
1141
-
1142
-
1143
-
1144
-
1145
-
1146
-
1147
-
1148
-
1149
-
1150
-
1151
-
1152
-
1153
-
1154
-
1155
-
1156
-
1157
-
1158
-
1159
-
1160
-
1161
-
1162
-
1163
-
1164
-
1165
-
1166
-
1167
-
1168
-
1169
-
1170
-
1171
-
1172
-
1173
-
1174
-
1175
-
1176
-
1177
-
1178
-
1179
-
1180
-
1181
-
1182
-
1183
-
1184
-
1185
-
1186
-
1187
-
1188
-
1189
-
1190
-
1191
-
1192
-
1193
-
1194
-
1195
-
1196
-
1197
-
1198
-
1199
-
1200
-
1201
-
1202
-
1203
-
1204
-
1205
-
1206
-
1207
-
1208
-
1209
-
1210
-
1211
-
1212
-
1213
-
1214
-
1215
-
1216
-
1217
-
1218
-
1219
-
1220
-
1221
-
1222
-
1223
-
1224
-
1225
-
1226
-
1227
-
1228
-
1229
-
1230
-
1231
-
1232
-
1233
-
1234
-
1235
-
1236
-
1237
-
1238
-
1239
-
1240
-
1241
-
1242
-
1243
-
1244
-
1245
-
1246
-
1247
-
1248
-
1249
-
1250
-
1251
-
1252
-
1253
-
1254
-
1255
-
1256
-
1257
-
1258
-
1259
-
1260
-
1261
-
1262
-
1263
-
1264
-
1265
-
1266
-
1267
-
1268
-
1269
-
1270
-
1271
-
1272
-
1273
-
1274
-
1275
-
1276
-
1277
-
1278
-
1279
-
1280
-
1281
-
1282
-
1283
-
1284
-
1285
-
1286
-
1287
-
1288
-
1289
-
1290
-
1291
-
1292
-
1293
-
1294
-
1295
-
1296
-
1297
-
1298
-
1299
-
1300
-
1301
-
1302
-
1303
-
1304
-
1305
-
1306
-
1307
-
1308
-
1309
-
1310
-
1311
-
1312
-
1313
-
1314
-
1315
-
1316
-
1317
-
1318
-
1319
-
1320
-
1321
-
1322
-
1323
-
1324
-
1325
-
1326
-
1327
-
1328
-
1329
-
1330
-
1331
-
1332
-
1333
-
1334
-
1335
-
1336
-
1337
-
1338
-
1339
-
1340
-
1341
-
1342
-
1343
-
1344
-
1345
-
1346
-
1347
-
1348
-
1349
-
1350
-
1351
-
1352
-
1353
-
1354
-
1355
-
1356
-
1357
-
1358
-
1359
-
1360
-
1361
-
1362
-
1363
-
1364
-
1365
-
1366
-
1367
-
1368
-
1369
-
1370
-
1371
-
1372
-
1373
-
1374
-
1375
-
1376
-
1377
-
1378
-
1379
-
1380
-
1381
-
1382
-
1383
-
1384
-
1385
-
1386
-
1387
-
1388
-
1389
-
1390
-
1391
-
1392
-
1393
-
1394
-
1395
-
1396
-
1397
-
1398
-
1399
-
1400
-
1401
-
1402
-
1403
-
1404
-
1405
-
1406
-
1407
-
1408
-
1409
-
1410
-
1411
-
1412
-
1413
-
1414
-
1415
-
1416
-
1417
-
1418
-
1419
-
1420
-
1421
-
1422
-
1423
-
1424
-
1425
-
1426
-
1427
-
1428
-
1429
-
1430
-
1431
-
1432
-
1433
-
1434
-
1435
-
1436
-
1437
-
1438
-
1439
-
1440
-
1441
-
1442
-
1443
-
1444
-
1445
-
1446
-
1447
-
1448
-
1449
-
1450
-
1451
-
1452
-
1453
-
1454
-
1455
-
1456
-
1457
-
1458
-
1459
-
1460
-
1461
-
1462
-
1463
-
1464
-
1465
-
1466
-
1467
-
1468
-
1469
-
1470
-
1471
-
1472
-
1473
-
1474
-
1475
-
1476
-
1477
-
1478
-
1479
-
1480
-
1481
-
1482
-
1483
-
1484
-
1485
-
1486
-
1487
-
1488
-
1489
-
1490
-
1491
-
1492
-
1493
-
1494
-
1495
-
1496
-
1497
-
1498
-
1499
-
1500
-
1501
-
1502
-
1503
-
1504
-
1505
-
1506
-
1507
-
1508
-
1509
-
1510
-
1511
-
1512
-
1513
-
1514
-
1515
-
1516
-
1517
-
1518
-
1519
-
1520
-
1521
-
1522
-
1523
-
1524
-
1525
-
1526
-
1527
-
1528
-
1529
-
1530
-
1531
-
1532
-
1533
-
1534
-
1535
-
1536
-
1537
-
1538
-
1539
-
1540
-
1541
-
1542
-
1543
-
1544
-
1545
-
1546
-
1547
-
1548
-
1549
-
1550
-
1551
-
1552
-
1553
-
1554
-
1555
-
1556
-
1557
-
1558
-
1559
-
1560
-
1561
-
1562
-
1563
-
1564
-
1565
-
1566
-
1567
-
1568
-
1569
-
1570
-
1571
-
1572
-
1573
-
1574
-
1575
-
1576
-
1577
-
1578
-
1579
-
1580
-
1581
-
1582
-
1583
-
1584
-
1585
-
1586
-
1587
-
1588
-
1589
-
1590
-
1591
-
1592
-
1593
-
1594
-
1595
-
1596
-
1597
-
1598
-
1599
-
1600
-
1601
-
1602
-
1603
-
1604
-
1605
-
1606
-
1607
-
1608
-
1609
-
1610
-
1611
-
1612
-
1613
-
1614
-
1615
-
1616
-
1617
-
1618
-
1619
-
1620
-
1621
-
1622
-
1623
-
1624
-
1625
-
1626
-
1627
-
1628
-
1629
-
1630
-
1631
-
1632
-
1633
-
1634
-
1635
-
1636
-
1637
-
1638
-
1639
-
1640
-
1641
-
1642
-
1643
-
1644
-
1645
-
1646
-
1647
-
1648
-
1649
-
1650
-
1651
-
1652
-
1653
-
1654
-
1655
-
1656
-
1657
-
1658
-
1659
-
1660
-
1661
-
1662
-
1663
-
1664
-
1665
-
1666
-
1667
-
1668
-
1669
-
1670
-
1671
-
1672
-
1673
-
1674
-
1675
-
1676
-
1677
-
1678
-
1679
-
1680
-
1681
-
1682
-
1683
-
1684
-
1685
-
1686
-
1687
-
1688
-
1689
-
1690
-
1691
-
1692
-
1693
-
1694
-
1695
-
1696
-
1697
-
1698
-
1699
-
1700
-
1701
-
1702
-
1703
-
1704
-
1705
-
1706
-
1707
-
1708
-
1709
-
1710
-
1711
-
1712
-
1713
-
1714
-
1715
-
1716
-
1717
-
1718
-
1719
-
1720
-
1721
-
1722
-
1723
-
1724
-
1725
-
1726
-
1727
-
1728
-
1729
-
1730
-
1731
-
1732
-
1733
-
1734
-
1735
-
1736
-
1737
-
1738
-
1739
-
1740
-
1741
-
1742
-
1743
-
1744
-
1745
-
1746
-
1747
-
1748
-
1749
-
1750
-
1751
-
1752
-
1753
-
1754
-
1755
-
1756
-
1757
-
1758
-
1759
-
1760
-
1761
-
1762
-
1763
-
1764
-
1765
-
1766
-
1767
-
1768
-
1769
-
1770
-
1771
-
1772
-
1773
-
1774
-
1775
-
1776
-
1777
-
1778
-
1779
-
1780
-
1781
-
1782
-
1783
-
1784
-
1785
-
1786
-
1787
-
1788
-
1789
-
1790
-
1791
-
1792
-
1793
-
1794
-
1795
-
1796
-
1797
-
1798
-
1799
-
1800
-
1801
-
1802
-
1803
-
1804
-
1805
-
1806
-
1807
-
1808
-
1809
-
1810
-
1811
-
1812
-
1813
-
1814
-
1815
-
1816
-
1817
-
1818
-
1819
-
1820
-
1821
-
1822
-
1823
-
1824
-
1825
-
1826
-
1827
-
1828
-
1829
-
1830
-
1831
-
1832
-
1833
-
1834
-
1835
-
1836
-
1837
-
1838
-
1839
-
1840
-
1841
-
1842
-
1843
-
1844
-
1845
-
1846
-
1847
-
1848
-
1849
-
1850
-
1851
-
1852
-
1853
-
1854
-
1855
-
1856
-
1857
-
1858
-
1859
-
1860
-
1861
-
1862
-
1863
-
1864
-
1865
-
1866
-
1867
-
1868
-
1869
-
1870
-
1871
-
1872
-
1873
-
1874
-
1875
-
1876
-
1877
-
1878
-
1879
-
1880
-
1881
-
1882
-
1883
-
1884
-
1885
-
1886
-
1887
-
1888
-
1889
-
1890
-
1891
-
1892
-
1893
-
1894
-
1895
-
1896
-
1897
-
1898
-
1899
-
1900
-
1901
-
1902
-
1903
-
1904
-
1905
-
1906
-
1907
-
1908
-
1909
-
1910
-
1911
-
1912
-
1913
-
1914
-
1915
-
1916
-
1917
-
1918
-
1919
-
1920
-
1921
-
1922
-
1923
-
1924
-
1925
-
1926
-
1927
-
1928
-
1929
-
1930
-
1931
-
1932
-
1933
-
1934
-
1935
-
1936
-
1937
-
1938
-
1939
-
1940
-
1941
-
1942
-
1943
-
1944
-
1945
-
1946
-
1947
-
1948
-
1949
-
1950
-
1951
-
1952
-
1953
-
1954
-
1955
-
1956
-
1957
-
1958
-
1959
-
1960
-
1961
-
1962
-
1963
-
1964
-
1965
-
1966
-
1967
-
1968
-
1969
-
1970
-
1971
-
1972
-
1973
-
1974
-
1975
-
1976
-
1977
-
1978
-
1979
-
1980
-
1981
-
1982
-
1983
-
1984
-
1985
-
1986
-
1987
-
1988
-
1989
-
1990
-
1991
-
1992
-
1993
-
1994
-
1995
-
1996
-
1997
-
1998
-
1999
-
2000
-
2001
-
2002
-
2003
-
2004
-
2005
-
2006
-
2007
-
2008
-
2009
-
2010
-
2011
-
2012
-
2013
-
2014
-
2015
-
2016
-
2017
-
2018
-
2019
-
2020
-
2021
-
2022
-
2023
-
2024
-
2025
-
2026
-
2027
-
2028
-
2029
-
2030
-
2031
-
2032
-
2033
-
2034
-
2035
-
2036
-
2037
-
2038
-
2039
-
2040
-
2041
-
2042
-
2043
-
2044
-
2045
-
2046
-
2047
-
2048
-
2049
-
2050
-
2051
-
2052
-
2053
-
2054
-
2055
-
2056
-
2057
-
2058
-
2059
-
2060
-
2061
-
2062
-
2063
-
2064
-
2065
-
2066
-
2067
-
2068
-
2069
-
2070
-
2071
-
2072
-
2073
-
2074
-
2075
-
2076
-
2077
-
2078
-
2079
-
2080
-
2081
-
2082
-
2083
-
2084
-
2085
-
2086
-
2087
-
2088
-
2089
-
2090
-
2091
-
2092
-
2093
-
2094
-
2095
-
2096
-
2097
-
2098
-
2099
-
2100
-
2101
-
2102
-
2103
-
2104
-
2105
-
2106
-
2107
-
2108
-
2109
-
2110
-
2111
-
2112
-
2113
-
2114
-
2115
-
2116
-
2117
-
2118
-
2119
-
2120
-
2121
-
2122
-
2123
-
2124
-
2125
-
2126
-
2127
-
2128
-
2129
-
2130
-
2131
-
2132
-
2133
-
2134
-
2135
-
2136
-
2137
-
2138
-
2139
-
2140
-
2141
-
2142
-
2143
-
2144
-
2145
-
2146
-
2147
-
2148
-
2149
-
2150
-
2151
-
2152
-
2153
-
2154
-
2155
-
2156
-
2157
-
2158
-
2159
-
2160
-
2161
-
2162
-
2163
-
2164
-
2165
-
2166
-
2167
-
2168
-
2169
-
2170
-
2171
-
2172
-
2173
-
2174
-
2175
-
2176
-
2177
-
2178
-
2179
-
2180
-
2181
-
2182
-
2183
-
2184
-
2185
-
2186
-
2187
-
2188
-
2189
-
2190
-
2191
-
2192
-
2193
-
2194
-
2195
-
2196
-
2197
-
2198
-
2199
-
2200
-
2201
-
2202
-
2203
-
2204
-
2205
-
2206
-
2207
-
2208
-
2209
-
2210
-
2211
-
2212
-
2213
-
2214
-
2215
-
2216
-
2217
-
2218
-
2219
-
2220
-
2221
-
2222
-
2223
-
2224
-
2225
-
2226
-
2227
-
2228
-
2229
-
2230
-
2231
-
2232
-
2233
-
2234
-
2235
-
2236
-
2237
-
2238
-
2239
-
2240
-
2241
-
2242
-
2243
-
2244
-
2245
-
2246
-
2247
-
2248
-
2249
-
2250
-
2251
-
2252
-
2253
-
2254
-
2255
-
2256
-
2257
-
2258
-
2259
-
2260
-
2261
-
2262
-
2263
-
2264
-
2265
-
2266
-
2267
-
2268
-
2269
-
2270
-
2271
-
2272
-
2273
-
2274
-
2275
-
2276
-
2277
-
2278
-
2279
-
2280
-
2281
-
2282
-
2283
-
2284
-
2285
-
2286
-
2287
-
2288
-
2289
-
2290
-
2291
-
2292
-
2293
-
2294
-
2295
// Copyright 2012-present Oliver Eilhard. All rights reserved.
// Use of this source code is governed by a MIT-license.
// See http://olivere.mit-license.org/license.txt for details.
package elastic
import (
"bytes"
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"net/http/httputil"
"net/url"
"os"
"runtime"
"strings"
"sync"
"time"
"github.com/pkg/errors"
"github.com/olivere/elastic/v7/config"
)
const (
// Version is the current version of Elastic.
Version = "7.0.25"
// DefaultURL is the default endpoint of Elasticsearch on the local machine.
// It is used e.g. when initializing a new Client without a specific URL.
DefaultURL = "http://127.0.0.1:9200"
// DefaultScheme is the default protocol scheme to use when sniffing
// the Elasticsearch cluster.
DefaultScheme = "http"
// DefaultHealthcheckEnabled specifies if healthchecks are enabled by default.
DefaultHealthcheckEnabled = true
// DefaultHealthcheckTimeoutStartup is the time the healthcheck waits
// for a response from Elasticsearch on startup, i.e. when creating a
// client. After the client is started, a shorter timeout is commonly used
// (its default is specified in DefaultHealthcheckTimeout).
DefaultHealthcheckTimeoutStartup = 5 * time.Second
// DefaultHealthcheckTimeout specifies the time a running client waits for
// a response from Elasticsearch. Notice that the healthcheck timeout
// when a client is created is larger by default (see DefaultHealthcheckTimeoutStartup).
DefaultHealthcheckTimeout = 1 * time.Second
// DefaultHealthcheckInterval is the default interval between
// two health checks of the nodes in the cluster.
DefaultHealthcheckInterval = 60 * time.Second
// DefaultSnifferEnabled specifies if the sniffer is enabled by default.
DefaultSnifferEnabled = true
// DefaultSnifferInterval is the interval between two sniffing procedures,
// i.e. the lookup of all nodes in the cluster and their addition/removal
// from the list of actual connections.
DefaultSnifferInterval = 15 * time.Minute
// DefaultSnifferTimeoutStartup is the default timeout for the sniffing
// process that is initiated while creating a new client. For subsequent
// sniffing processes, DefaultSnifferTimeout is used (by default).
DefaultSnifferTimeoutStartup = 5 * time.Second
// DefaultSnifferTimeout is the default timeout after which the
// sniffing process times out. Notice that for the initial sniffing
// process, DefaultSnifferTimeoutStartup is used.
DefaultSnifferTimeout = 2 * time.Second
// DefaultSendGetBodyAs is the HTTP method to use when elastic is sending
// a GET request with a body.
DefaultSendGetBodyAs = "GET"
// DefaultGzipEnabled specifies if gzip compression is enabled by default.
DefaultGzipEnabled = false
// off is used to disable timeouts.
off = -1 * time.Second
)
var (
// ErrNoClient is raised when no Elasticsearch node is available.
ErrNoClient = errors.New("no Elasticsearch node available")
// ErrRetry is raised when a request cannot be executed after the configured
// number of retries.
ErrRetry = errors.New("cannot connect after several retries")
// ErrTimeout is raised when a request timed out, e.g. when WaitForStatus
// didn't return in time.
ErrTimeout = errors.New("timeout")
// noRetries is a retrier that does not retry.
noRetries = NewStopRetrier()
// noDeprecationLog is a no-op for logging deprecations.
noDeprecationLog = func(*http.Request, *http.Response) {}
)
// Doer is an interface to perform HTTP requests.
// It can be used for mocking.
type Doer interface {
Do(*http.Request) (*http.Response, error)
}
// ClientOptionFunc is a function that configures a Client.
// It is used in NewClient.
type ClientOptionFunc func(*Client) error
// Client is an Elasticsearch client. Create one by calling NewClient.
type Client struct {
c Doer // e.g. a net/*http.Client to use for requests
connsMu sync.RWMutex // connsMu guards the next block
conns []*conn // all connections
cindex int // index into conns
mu sync.RWMutex // guards the next block
urls []string // set of URLs passed initially to the client
running bool // true if the client's background processes are running
errorlog Logger // error log for critical messages
infolog Logger // information log for e.g. response times
tracelog Logger // trace log for debugging
deprecationlog func(*http.Request, *http.Response)
scheme string // http or https
healthcheckEnabled bool // healthchecks enabled or disabled
healthcheckTimeoutStartup time.Duration // time the healthcheck waits for a response from Elasticsearch on startup
healthcheckTimeout time.Duration // time the healthcheck waits for a response from Elasticsearch
healthcheckInterval time.Duration // interval between healthchecks
healthcheckStop chan bool // notify healthchecker to stop, and notify back
snifferEnabled bool // sniffer enabled or disabled
snifferTimeoutStartup time.Duration // time the sniffer waits for a response from nodes info API on startup
snifferTimeout time.Duration // time the sniffer waits for a response from nodes info API
snifferInterval time.Duration // interval between sniffing
snifferCallback SnifferCallback // callback to modify the sniffing decision
snifferStop chan bool // notify sniffer to stop, and notify back
decoder Decoder // used to decode data sent from Elasticsearch
basicAuthUsername string // username for HTTP Basic Auth
basicAuthPassword string // password for HTTP Basic Auth
sendGetBodyAs string // override for when sending a GET with a body
gzipEnabled bool // gzip compression enabled or disabled (default)
requiredPlugins []string // list of required plugins
retrier Retrier // strategy for retries
retryStatusCodes []int // HTTP status codes where to retry automatically (with retrier)
headers http.Header // a list of default headers to add to each request
}
// NewClient creates a new client to work with Elasticsearch.
//
// NewClient, by default, is meant to be long-lived and shared across
// your application. If you need a short-lived client, e.g. for request-scope,
// consider using NewSimpleClient instead.
//
// The caller can configure the new client by passing configuration options
// to the func.
//
// Example:
//
// client, err := elastic.NewClient(
// elastic.SetURL("http://127.0.0.1:9200", "http://127.0.0.1:9201"),
// elastic.SetBasicAuth("user", "secret"))
//
// If no URL is configured, Elastic uses DefaultURL by default.
//
// If the sniffer is enabled (the default), the new client then sniffes
// the cluster via the Nodes Info API
// (see https://www.elastic.co/guide/en/elasticsearch/reference/7.0/cluster-nodes-info.html#cluster-nodes-info).
// It uses the URLs specified by the caller. The caller is responsible
// to only pass a list of URLs of nodes that belong to the same cluster.
// This sniffing process is run on startup and periodically.
// Use SnifferInterval to set the interval between two sniffs (default is
// 15 minutes). In other words: By default, the client will find new nodes
// in the cluster and remove those that are no longer available every
// 15 minutes. Disable the sniffer by passing SetSniff(false) to NewClient.
//
// The list of nodes found in the sniffing process will be used to make
// connections to the REST API of Elasticsearch. These nodes are also
// periodically checked in a shorter time frame. This process is called
// a health check. By default, a health check is done every 60 seconds.
// You can set a shorter or longer interval by SetHealthcheckInterval.
// Disabling health checks is not recommended, but can be done by
// SetHealthcheck(false).
//
// Connections are automatically marked as dead or healthy while
// making requests to Elasticsearch. When a request fails, Elastic will
// call into the Retry strategy which can be specified with SetRetry.
// The Retry strategy is also responsible for handling backoff i.e. the time
// to wait before starting the next request. There are various standard
// backoff implementations, e.g. ExponentialBackoff or SimpleBackoff.
// Retries are disabled by default.
//
// If no HttpClient is configured, then http.DefaultClient is used.
// You can use your own http.Client with some http.Transport for
// advanced scenarios.
//
// An error is also returned when some configuration option is invalid or
// the new client cannot sniff the cluster (if enabled).
func NewClient(options ...ClientOptionFunc) (*Client, error) {
return DialContext(context.Background(), options...)
}
// NewClientFromConfig initializes a client from a configuration.
func NewClientFromConfig(cfg *config.Config) (*Client, error) {
options, err := configToOptions(cfg)
if err != nil {
return nil, err
}
return DialContext(context.Background(), options...)
}
// NewSimpleClient creates a new short-lived Client that can be used in
// use cases where you need e.g. one client per request.
//
// While NewClient by default sets up e.g. periodic health checks
// and sniffing for new nodes in separate goroutines, NewSimpleClient does
// not and is meant as a simple replacement where you don't need all the
// heavy lifting of NewClient.
//
// NewSimpleClient does the following by default: First, all health checks
// are disabled, including timeouts and periodic checks. Second, sniffing
// is disabled, including timeouts and periodic checks. The number of retries
// is set to 1. NewSimpleClient also does not start any goroutines.
//
// Notice that you can still override settings by passing additional options,
// just like with NewClient.
func NewSimpleClient(options ...ClientOptionFunc) (*Client, error) {
c := &Client{
c: http.DefaultClient,
conns: make([]*conn, 0),
cindex: -1,
scheme: DefaultScheme,
decoder: &DefaultDecoder{},
healthcheckEnabled: false,
healthcheckTimeoutStartup: off,
healthcheckTimeout: off,
healthcheckInterval: off,
healthcheckStop: make(chan bool),
snifferEnabled: false,
snifferTimeoutStartup: off,
snifferTimeout: off,
snifferInterval: off,
snifferCallback: nopSnifferCallback,
snifferStop: make(chan bool),
sendGetBodyAs: DefaultSendGetBodyAs,
gzipEnabled: DefaultGzipEnabled,
retrier: noRetries, // no retries by default
retryStatusCodes: nil, // no automatic retries for specific HTTP status codes
deprecationlog: noDeprecationLog,
}
// Run the options on it
for _, option := range options {
if err := option(c); err != nil {
return nil, err
}
}
// Use a default URL and normalize them
if len(c.urls) == 0 {
c.urls = []string{DefaultURL}
}
c.urls = canonicalize(c.urls...)
// If the URLs have auth info, use them here as an alternative to SetBasicAuth
if c.basicAuthUsername == "" && c.basicAuthPassword == "" {
for _, urlStr := range c.urls {
u, err := url.Parse(urlStr)
if err == nil && u.User != nil {
c.basicAuthUsername = u.User.Username()
c.basicAuthPassword, _ = u.User.Password()
break
}
}
}
for _, url := range c.urls {
c.conns = append(c.conns, newConn(url, url))
}
// Ensure that we have at least one connection available
if err := c.mustActiveConn(); err != nil {
return nil, err
}
// Check the required plugins
for _, plugin := range c.requiredPlugins {
found, err := c.HasPlugin(plugin)
if err != nil {
return nil, err
}
if !found {
return nil, fmt.Errorf("elastic: plugin %s not found", plugin)
}
}
c.mu.Lock()
c.running = true
c.mu.Unlock()
return c, nil
}
// Dial will call DialContext with a background context.
func Dial(options ...ClientOptionFunc) (*Client, error) {
return DialContext(context.Background(), options...)
}
// DialContext will connect to Elasticsearch, just like NewClient does.
//
// The context is honoured in terms of e.g. cancellation.
func DialContext(ctx context.Context, options ...ClientOptionFunc) (*Client, error) {
// Set up the client
c := &Client{
c: http.DefaultClient,
conns: make([]*conn, 0),
cindex: -1,
scheme: DefaultScheme,
decoder: &DefaultDecoder{},
healthcheckEnabled: DefaultHealthcheckEnabled,
healthcheckTimeoutStartup: DefaultHealthcheckTimeoutStartup,
healthcheckTimeout: DefaultHealthcheckTimeout,
healthcheckInterval: DefaultHealthcheckInterval,
healthcheckStop: make(chan bool),
snifferEnabled: DefaultSnifferEnabled,
snifferTimeoutStartup: DefaultSnifferTimeoutStartup,
snifferTimeout: DefaultSnifferTimeout,
snifferInterval: DefaultSnifferInterval,
snifferCallback: nopSnifferCallback,
snifferStop: make(chan bool),
sendGetBodyAs: DefaultSendGetBodyAs,
gzipEnabled: DefaultGzipEnabled,
retrier: noRetries, // no retries by default
retryStatusCodes: nil, // no automatic retries for specific HTTP status codes
deprecationlog: noDeprecationLog,
}
// Run the options on it
for _, option := range options {
if err := option(c); err != nil {
return nil, err
}
}
// Use a default URL and normalize them
if len(c.urls) == 0 {
c.urls = []string{DefaultURL}
}
c.urls = canonicalize(c.urls...)
// If the URLs have auth info, use them here as an alternative to SetBasicAuth
if c.basicAuthUsername == "" && c.basicAuthPassword == "" {
for _, urlStr := range c.urls {
u, err := url.Parse(urlStr)
if err == nil && u.User != nil {
c.basicAuthUsername = u.User.Username()
c.basicAuthPassword, _ = u.User.Password()
break
}
}
}
// Check if we can make a request to any of the specified URLs
if c.healthcheckEnabled {
if err := c.startupHealthcheck(ctx, c.healthcheckTimeoutStartup); err != nil {
return nil, err
}
}
if c.snifferEnabled {
// Sniff the cluster initially
if err := c.sniff(ctx, c.snifferTimeoutStartup); err != nil {
return nil, err
}
} else {
// Do not sniff the cluster initially. Use the provided URLs instead.
for _, url := range c.urls {
c.conns = append(c.conns, newConn(url, url))
}
}
if c.healthcheckEnabled {
// Perform an initial health check
c.healthcheck(ctx, c.healthcheckTimeoutStartup, true)
}
// Ensure that we have at least one connection available
if err := c.mustActiveConn(); err != nil {
return nil, err
}
// Check the required plugins
for _, plugin := range c.requiredPlugins {
found, err := c.HasPlugin(plugin)
if err != nil {
return nil, err
}
if !found {
return nil, fmt.Errorf("elastic: plugin %s not found", plugin)
}
}
if c.snifferEnabled {
go c.sniffer() // periodically update cluster information
}
if c.healthcheckEnabled {
go c.healthchecker() // start goroutine periodically ping all nodes of the cluster
}
c.mu.Lock()
c.running = true
c.mu.Unlock()
return c, nil
}
// DialWithConfig will use the configuration settings parsed from config package
// to connect to Elasticsearch.
//
// The context is honoured in terms of e.g. cancellation.
func DialWithConfig(ctx context.Context, cfg *config.Config) (*Client, error) {
options, err := configToOptions(cfg)
if err != nil {
return nil, err
}
return DialContext(ctx, options...)
}
func configToOptions(cfg *config.Config) ([]ClientOptionFunc, error) {
var options []ClientOptionFunc
if cfg != nil {
if cfg.URL != "" {
options = append(options, SetURL(cfg.URL))
}
if cfg.Errorlog != "" {
f, err := os.OpenFile(cfg.Errorlog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return nil, errors.Wrap(err, "unable to initialize error log")
}
l := log.New(f, "", 0)
options = append(options, SetErrorLog(l))
}
if cfg.Tracelog != "" {
f, err := os.OpenFile(cfg.Tracelog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return nil, errors.Wrap(err, "unable to initialize trace log")
}
l := log.New(f, "", 0)
options = append(options, SetTraceLog(l))
}
if cfg.Infolog != "" {
f, err := os.OpenFile(cfg.Infolog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return nil, errors.Wrap(err, "unable to initialize info log")
}
l := log.New(f, "", 0)
options = append(options, SetInfoLog(l))
}
if cfg.Username != "" || cfg.Password != "" {
options = append(options, SetBasicAuth(cfg.Username, cfg.Password))
}
if cfg.Sniff != nil {
options = append(options, SetSniff(*cfg.Sniff))
}
if cfg.Healthcheck != nil {
options = append(options, SetHealthcheck(*cfg.Healthcheck))
}
}
return options, nil
}
// SetHttpClient can be used to specify the http.Client to use when making
// HTTP requests to Elasticsearch.
func SetHttpClient(httpClient Doer) ClientOptionFunc {
return func(c *Client) error {
if httpClient != nil {
c.c = httpClient
} else {
c.c = http.DefaultClient
}
return nil
}
}
// SetBasicAuth can be used to specify the HTTP Basic Auth credentials to
// use when making HTTP requests to Elasticsearch.
func SetBasicAuth(username, password string) ClientOptionFunc {
return func(c *Client) error {
c.basicAuthUsername = username
c.basicAuthPassword = password
return nil
}
}
// SetURL defines the URL endpoints of the Elasticsearch nodes. Notice that
// when sniffing is enabled, these URLs are used to initially sniff the
// cluster on startup.
func SetURL(urls ...string) ClientOptionFunc {
return func(c *Client) error {
switch len(urls) {
case 0:
c.urls = []string{DefaultURL}
default:
c.urls = urls
}
// Check URLs
for _, urlStr := range c.urls {
if _, err := url.Parse(urlStr); err != nil {
return err
}
}
return nil
}
}
// SetScheme sets the HTTP scheme to look for when sniffing (http or https).
// This is http by default.
func SetScheme(scheme string) ClientOptionFunc {
return func(c *Client) error {
c.scheme = scheme
return nil
}
}
// SetSniff enables or disables the sniffer (enabled by default).
func SetSniff(enabled bool) ClientOptionFunc {
return func(c *Client) error {
c.snifferEnabled = enabled
return nil
}
}
// SetSnifferTimeoutStartup sets the timeout for the sniffer that is used
// when creating a new client. The default is 5 seconds. Notice that the
// timeout being used for subsequent sniffing processes is set with
// SetSnifferTimeout.
func SetSnifferTimeoutStartup(timeout time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.snifferTimeoutStartup = timeout
return nil
}
}
// SetSnifferTimeout sets the timeout for the sniffer that finds the
// nodes in a cluster. The default is 2 seconds. Notice that the timeout
// used when creating a new client on startup is usually greater and can
// be set with SetSnifferTimeoutStartup.
func SetSnifferTimeout(timeout time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.snifferTimeout = timeout
return nil
}
}
// SetSnifferInterval sets the interval between two sniffing processes.
// The default interval is 15 minutes.
func SetSnifferInterval(interval time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.snifferInterval = interval
return nil
}
}
// SnifferCallback defines the protocol for sniffing decisions.
type SnifferCallback func(*NodesInfoNode) bool
// nopSnifferCallback is the default sniffer callback: It accepts
// all nodes the sniffer finds.
var nopSnifferCallback = func(*NodesInfoNode) bool { return true }
// SetSnifferCallback allows the caller to modify sniffer decisions.
// When setting the callback, the given SnifferCallback is called for
// each (healthy) node found during the sniffing process.
// If the callback returns false, the node is ignored: No requests
// are routed to it.
func SetSnifferCallback(f SnifferCallback) ClientOptionFunc {
return func(c *Client) error {
if f != nil {
c.snifferCallback = f
}
return nil
}
}
// SetHealthcheck enables or disables healthchecks (enabled by default).
func SetHealthcheck(enabled bool) ClientOptionFunc {
return func(c *Client) error {
c.healthcheckEnabled = enabled
return nil
}
}
// SetHealthcheckTimeoutStartup sets the timeout for the initial health check.
// The default timeout is 5 seconds (see DefaultHealthcheckTimeoutStartup).
// Notice that timeouts for subsequent health checks can be modified with
// SetHealthcheckTimeout.
func SetHealthcheckTimeoutStartup(timeout time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.healthcheckTimeoutStartup = timeout
return nil
}
}
// SetHealthcheckTimeout sets the timeout for periodic health checks.
// The default timeout is 1 second (see DefaultHealthcheckTimeout).
// Notice that a different (usually larger) timeout is used for the initial
// healthcheck, which is initiated while creating a new client.
// The startup timeout can be modified with SetHealthcheckTimeoutStartup.
func SetHealthcheckTimeout(timeout time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.healthcheckTimeout = timeout
return nil
}
}
// SetHealthcheckInterval sets the interval between two health checks.
// The default interval is 60 seconds.
func SetHealthcheckInterval(interval time.Duration) ClientOptionFunc {
return func(c *Client) error {
c.healthcheckInterval = interval
return nil
}
}
// SetMaxRetries sets the maximum number of retries before giving up when
// performing a HTTP request to Elasticsearch.
//
// Deprecated: Replace with a Retry implementation.
func SetMaxRetries(maxRetries int) ClientOptionFunc {
return func(c *Client) error {
if maxRetries < 0 {
return errors.New("MaxRetries must be greater than or equal to 0")
} else if maxRetries == 0 {
c.retrier = noRetries
} else {
// Create a Retrier that will wait for 100ms (+/- jitter) between requests.
// This resembles the old behavior with maxRetries.
ticks := make([]int, maxRetries)
for i := 0; i < len(ticks); i++ {
ticks[i] = 100
}
backoff := NewSimpleBackoff(ticks...)
c.retrier = NewBackoffRetrier(backoff)
}
return nil
}
}
// SetGzip enables or disables gzip compression (disabled by default).
func SetGzip(enabled bool) ClientOptionFunc {
return func(c *Client) error {
c.gzipEnabled = enabled
return nil
}
}
// SetDecoder sets the Decoder to use when decoding data from Elasticsearch.
// DefaultDecoder is used by default.
func SetDecoder(decoder Decoder) ClientOptionFunc {
return func(c *Client) error {
if decoder != nil {
c.decoder = decoder
} else {
c.decoder = &DefaultDecoder{}
}
return nil
}
}
// SetRequiredPlugins can be used to indicate that some plugins are required
// before a Client will be created.
func SetRequiredPlugins(plugins ...string) ClientOptionFunc {
return func(c *Client) error {
if c.requiredPlugins == nil {
c.requiredPlugins = make([]string, 0)
}
c.requiredPlugins = append(c.requiredPlugins, plugins...)
return nil
}
}
// SetErrorLog sets the logger for critical messages like nodes joining
// or leaving the cluster or failing requests. It is nil by default.
func SetErrorLog(logger Logger) ClientOptionFunc {
return func(c *Client) error {
c.errorlog = logger
return nil
}
}
// SetInfoLog sets the logger for informational messages, e.g. requests
// and their response times. It is nil by default.
func SetInfoLog(logger Logger) ClientOptionFunc {
return func(c *Client) error {
c.infolog = logger
return nil
}
}
// SetTraceLog specifies the log.Logger to use for output of HTTP requests
// and responses which is helpful during debugging. It is nil by default.
func SetTraceLog(logger Logger) ClientOptionFunc {
return func(c *Client) error {
c.tracelog = logger
return nil
}
}
// SetSendGetBodyAs specifies the HTTP method to use when sending a GET request
// with a body. It is GET by default.
func SetSendGetBodyAs(httpMethod string) ClientOptionFunc {
return func(c *Client) error {
c.sendGetBodyAs = httpMethod
return nil
}
}
// SetRetrier specifies the retry strategy that handles errors during
// HTTP request/response with Elasticsearch.
func SetRetrier(retrier Retrier) ClientOptionFunc {
return func(c *Client) error {
if retrier == nil {
retrier = noRetries // no retries by default
}
c.retrier = retrier
return nil
}
}
// SetRetryStatusCodes specifies the HTTP status codes where the client
// will retry automatically. Notice that retries call the specified retrier,
// so calling SetRetryStatusCodes without setting a Retrier won't do anything
// for retries.
func SetRetryStatusCodes(statusCodes ...int) ClientOptionFunc {
return func(c *Client) error {
c.retryStatusCodes = statusCodes
return nil
}
}
// SetHeaders adds a list of default HTTP headers that will be added to
// each requests executed by PerformRequest.
func SetHeaders(headers http.Header) ClientOptionFunc {
return func(c *Client) error {
c.headers = headers
return nil
}
}
// String returns a string representation of the client status.
func (c *Client) String() string {
c.connsMu.Lock()
conns := c.conns
c.connsMu.Unlock()
var buf bytes.Buffer
for i, conn := range conns {
if i > 0 {
buf.WriteString(", ")
}
buf.WriteString(conn.String())
}
return buf.String()
}
// IsRunning returns true if the background processes of the client are
// running, false otherwise.
func (c *Client) IsRunning() bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.running
}
// Start starts the background processes like sniffing the cluster and
// periodic health checks. You don't need to run Start when creating a
// client with NewClient; the background processes are run by default.
//
// If the background processes are already running, this is a no-op.
func (c *Client) Start() {
c.mu.RLock()
if c.running {
c.mu.RUnlock()
return
}
c.mu.RUnlock()
if c.snifferEnabled {
go c.sniffer()
}
if c.healthcheckEnabled {
go c.healthchecker()
}
c.mu.Lock()
c.running = true
c.mu.Unlock()
c.infof("elastic: client started")
}
// Stop stops the background processes that the client is running,
// i.e. sniffing the cluster periodically and running health checks
// on the nodes.
//
// If the background processes are not running, this is a no-op.
func (c *Client) Stop() {
c.mu.RLock()
if !c.running {
c.mu.RUnlock()
return
}
c.mu.RUnlock()
if c.healthcheckEnabled {
c.healthcheckStop <- true
<-c.healthcheckStop
}
if c.snifferEnabled {
c.snifferStop <- true
<-c.snifferStop
}
c.mu.Lock()
c.running = false
c.mu.Unlock()
c.infof("elastic: client stopped")
}
// errorf logs to the error log.
func (c *Client) errorf(format string, args ...interface{}) {
if c.errorlog != nil {
c.errorlog.Printf(format, args...)
}
}
// infof logs informational messages.
func (c *Client) infof(format string, args ...interface{}) {
if c.infolog != nil {
c.infolog.Printf(format, args...)
}
}
// tracef logs to the trace log.
func (c *Client) tracef(format string, args ...interface{}) {
if c.tracelog != nil {
c.tracelog.Printf(format, args...)
}
}
// dumpRequest dumps the given HTTP request to the trace log.
func (c *Client) dumpRequest(r *http.Request) {
if c.tracelog != nil {
out, err := httputil.DumpRequestOut(r, true)
if err == nil {
c.tracef("%s\n", string(out))
}
}
}
// dumpResponse dumps the given HTTP response to the trace log.
func (c *Client) dumpResponse(resp *http.Response) {
if c.tracelog != nil {
out, err := httputil.DumpResponse(resp, true)
if err == nil {
c.tracef("%s\n", string(out))
}
}
}
// sniffer periodically runs sniff.
func (c *Client) sniffer() {
c.mu.RLock()
timeout := c.snifferTimeout
interval := c.snifferInterval
c.mu.RUnlock()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-c.snifferStop:
// we are asked to stop, so we signal back that we're stopping now
c.snifferStop <- true
return
case <-ticker.C:
c.sniff(context.Background(), timeout)
}
}
}
// sniff uses the Node Info API to return the list of nodes in the cluster.
// It uses the list of URLs passed on startup plus the list of URLs found
// by the preceding sniffing process (if sniffing is enabled).
//
// If sniffing is disabled, this is a no-op.
func (c *Client) sniff(parentCtx context.Context, timeout time.Duration) error {
c.mu.RLock()
if !c.snifferEnabled {
c.mu.RUnlock()
return nil
}
// Use all available URLs provided to sniff the cluster.
var urls []string
urlsMap := make(map[string]bool)
// Add all URLs provided on startup
for _, url := range c.urls {
urlsMap[url] = true
urls = append(urls, url)
}
c.mu.RUnlock()
// Add all URLs found by sniffing
c.connsMu.RLock()
for _, conn := range c.conns {
if !conn.IsDead() {
url := conn.URL()
if _, found := urlsMap[url]; !found {
urls = append(urls, url)
}
}
}
c.connsMu.RUnlock()
if len(urls) == 0 {
return errors.Wrap(ErrNoClient, "no URLs found")
}
// Start sniffing on all found URLs
ch := make(chan []*conn, len(urls))
ctx, cancel := context.WithTimeout(parentCtx, timeout)
defer cancel()
for _, url := range urls {
go func(url string) { ch <- c.sniffNode(ctx, url) }(url)
}
// Wait for the results to come back, or the process times out.
for {
select {
case conns := <-ch:
if len(conns) > 0 {
c.updateConns(conns)
return nil
}
case <-ctx.Done():
if err := ctx.Err(); err != nil {
switch {
case IsContextErr(err):
return err
}
return errors.Wrapf(ErrNoClient, "sniff timeout: %v", err)
}
// We get here if no cluster responds in time
return errors.Wrap(ErrNoClient, "sniff timeout")
}
}
}
// sniffNode sniffs a single node. This method is run as a goroutine
// in sniff. If successful, it returns the list of node URLs extracted
// from the result of calling Nodes Info API. Otherwise, an empty array
// is returned.
func (c *Client) sniffNode(ctx context.Context, url string) []*conn {
var nodes []*conn
// Call the Nodes Info API at /_nodes/http
req, err := NewRequest("GET", url+"/_nodes/http")
if err != nil {
return nodes
}
c.mu.RLock()
if c.basicAuthUsername != "" || c.basicAuthPassword != "" {
req.SetBasicAuth(c.basicAuthUsername, c.basicAuthPassword)
}
c.mu.RUnlock()
if req.Header.Get("User-Agent") == "" {
req.Header.Add("User-Agent", "elastic/"+Version+" ("+runtime.GOOS+"-"+runtime.GOARCH+")")
}
res, err := c.c.Do((*http.Request)(req).WithContext(ctx))
if err != nil {
return nodes
}
defer res.Body.Close()
var info NodesInfoResponse
if err := json.NewDecoder(res.Body).Decode(&info); err == nil {
if len(info.Nodes) > 0 {
for nodeID, node := range info.Nodes {
if c.snifferCallback(node) {
if node.HTTP != nil && len(node.HTTP.PublishAddress) > 0 {
url := c.extractHostname(c.scheme, node.HTTP.PublishAddress)
if url != "" {
nodes = append(nodes, newConn(nodeID, url))
}
}
}
}
}
}
return nodes
}
// extractHostname returns the URL from the http.publish_address setting.
func (c *Client) extractHostname(scheme, address string) string {
var (
host string
port string
addrs = strings.Split(address, "/")
ports = strings.Split(address, ":")
)
if len(addrs) > 1 {
host = addrs[0]
} else {
host = strings.Split(addrs[0], ":")[0]
}
port = ports[len(ports)-1]
return fmt.Sprintf("%s://%s:%s", scheme, host, port)
}
// updateConns updates the clients' connections with new information
// gather by a sniff operation.
func (c *Client) updateConns(conns []*conn) {
c.connsMu.Lock()
// Build up new connections:
// If we find an existing connection, use that (including no. of failures etc.).
// If we find a new connection, add it.
var newConns []*conn
for _, conn := range conns {
var found bool
for _, oldConn := range c.conns {
// Notice that e.g. in a Kubernetes cluster the NodeID might be
// stable while the URL has changed.
if oldConn.NodeID() == conn.NodeID() && oldConn.URL() == conn.URL() {
// Take over the old connection
newConns = append(newConns, oldConn)
found = true
break
}
}
if !found {
// New connection didn't exist, so add it to our list of new conns.
c.infof("elastic: %s joined the cluster", conn.URL())
newConns = append(newConns, conn)
}
}
c.conns = newConns
c.cindex = -1
c.connsMu.Unlock()
}
// healthchecker periodically runs healthcheck.
func (c *Client) healthchecker() {
c.mu.RLock()
timeout := c.healthcheckTimeout
interval := c.healthcheckInterval
c.mu.RUnlock()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-c.healthcheckStop:
// we are asked to stop, so we signal back that we're stopping now
c.healthcheckStop <- true
return
case <-ticker.C:
c.healthcheck(context.Background(), timeout, false)
}
}
}
// healthcheck does a health check on all nodes in the cluster. Depending on
// the node state, it marks connections as dead, sets them alive etc.
// If healthchecks are disabled and force is false, this is a no-op.
// The timeout specifies how long to wait for a response from Elasticsearch.
func (c *Client) healthcheck(parentCtx context.Context, timeout time.Duration, force bool) {
c.mu.RLock()
if !c.healthcheckEnabled && !force {
c.mu.RUnlock()
return
}
headers := c.headers
basicAuth := c.basicAuthUsername != "" || c.basicAuthPassword != ""
basicAuthUsername := c.basicAuthUsername
basicAuthPassword := c.basicAuthPassword
c.mu.RUnlock()
c.connsMu.RLock()
conns := c.conns
c.connsMu.RUnlock()
for _, conn := range conns {
// Run the HEAD request against ES with a timeout
ctx, cancel := context.WithTimeout(parentCtx, timeout)
defer cancel()
// Goroutine executes the HTTP request, returns an error and sets status
var status int
errc := make(chan error, 1)
go func(url string) {
req, err := NewRequest("HEAD", url)
if err != nil {
errc <- err
return
}
if basicAuth {
req.SetBasicAuth(basicAuthUsername, basicAuthPassword)
}
if len(headers) > 0 {
for key, values := range headers {
for _, v := range values {
req.Header.Add(key, v)
}
}
}
if req.Header.Get("User-Agent") == "" {
req.Header.Add("User-Agent", "elastic/"+Version+" ("+runtime.GOOS+"-"+runtime.GOARCH+")")
}
res, err := c.c.Do((*http.Request)(req).WithContext(ctx))
if res != nil {
status = res.StatusCode
if res.Body != nil {
res.Body.Close()
}
}
errc <- err
}(conn.URL())
// Wait for the Goroutine (or its timeout)
select {
case <-ctx.Done(): // timeout
c.errorf("elastic: %s is dead", conn.URL())
conn.MarkAsDead()
case err := <-errc:
if err != nil {
c.errorf("elastic: %s is dead", conn.URL())
conn.MarkAsDead()
break
}
if status >= 200 && status < 300 {
conn.MarkAsAlive()
} else {
conn.MarkAsDead()
c.errorf("elastic: %s is dead [status=%d]", conn.URL(), status)
}
}
}
}
// startupHealthcheck is used at startup to check if the server is available
// at all.
func (c *Client) startupHealthcheck(parentCtx context.Context, timeout time.Duration) error {
c.mu.Lock()
urls := c.urls
headers := c.headers
basicAuth := c.basicAuthUsername != "" || c.basicAuthPassword != ""
basicAuthUsername := c.basicAuthUsername
basicAuthPassword := c.basicAuthPassword
c.mu.Unlock()
// If we don't get a connection after "timeout", we bail.
var lastErr error
start := time.Now()
done := false
for !done {
for _, url := range urls {
req, err := http.NewRequest("HEAD", url, nil)
if err != nil {
return err
}
if basicAuth {
req.SetBasicAuth(basicAuthUsername, basicAuthPassword)
}
if len(headers) > 0 {
for key, values := range headers {
for _, v := range values {
req.Header.Add(key, v)
}
}
}
ctx, cancel := context.WithTimeout(parentCtx, timeout)
defer cancel()
req = req.WithContext(ctx)
res, err := c.c.Do(req)
if err != nil {
lastErr = err
} else if res.StatusCode >= 200 && res.StatusCode < 300 {
return nil
} else if res.StatusCode == http.StatusUnauthorized {
lastErr = &Error{Status: res.StatusCode}
}
}
select {
case <-parentCtx.Done():
lastErr = parentCtx.Err()
done = true
case <-time.After(1 * time.Second):
if time.Since(start) > timeout {
done = true
}
}
}
if lastErr != nil {
if IsContextErr(lastErr) || IsUnauthorized(lastErr) {
return lastErr
}
return errors.Wrapf(ErrNoClient, "health check timeout: %v", lastErr)
}
return errors.Wrap(ErrNoClient, "health check timeout")
}
// next returns the next available connection, or ErrNoClient.
func (c *Client) next() (*conn, error) {
// We do round-robin here.
// TODO(oe) This should be a pluggable strategy, like the Selector in the official clients.
c.connsMu.Lock()
defer c.connsMu.Unlock()
i := 0
numConns := len(c.conns)
for {
i++
if i > numConns {
break // we visited all conns: they all seem to be dead
}
c.cindex++
if c.cindex >= numConns {
c.cindex = 0
}
conn := c.conns[c.cindex]
if !conn.IsDead() {
return conn, nil
}
}
// We have a deadlock here: All nodes are marked as dead.
// If sniffing is disabled, connections will never be marked alive again.
// So we are marking them as alive--if sniffing is disabled.
// They'll then be picked up in the next call to PerformRequest.
if !c.snifferEnabled {
c.errorf("elastic: all %d nodes marked as dead; resurrecting them to prevent deadlock", len(c.conns))
for _, conn := range c.conns {
conn.MarkAsAlive()
}
}
// We tried hard, but there is no node available
return nil, errors.Wrap(ErrNoClient, "no available connection")
}
// mustActiveConn returns nil if there is an active connection,
// otherwise ErrNoClient is returned.
func (c *Client) mustActiveConn() error {
c.connsMu.Lock()
defer c.connsMu.Unlock()
for _, c := range c.conns {
if !c.IsDead() {
return nil
}
}
return errors.Wrap(ErrNoClient, "no active connection found")
}
// -- PerformRequest --
// PerformRequestOptions must be passed into PerformRequest.
type PerformRequestOptions struct {
Method string
Path string
Params url.Values
Body interface{}
ContentType string
IgnoreErrors []int
Retrier Retrier
RetryStatusCodes []int
Headers http.Header
MaxResponseSize int64
Stream bool
}
// PerformRequest does a HTTP request to Elasticsearch.
// It returns a response (which might be nil) and an error on failure.
//
// Optionally, a list of HTTP error codes to ignore can be passed.
// This is necessary for services that expect e.g. HTTP status 404 as a
// valid outcome (Exists, IndicesExists, IndicesTypeExists).
//
// If Stream is set, the returned BodyReader field must be closed, even
// if PerformRequest returns an error.
func (c *Client) PerformRequest(ctx context.Context, opt PerformRequestOptions) (*Response, error) {
start := time.Now().UTC()
c.mu.RLock()
timeout := c.healthcheckTimeout
basicAuth := c.basicAuthUsername != "" || c.basicAuthPassword != ""
basicAuthUsername := c.basicAuthUsername
basicAuthPassword := c.basicAuthPassword
sendGetBodyAs := c.sendGetBodyAs
gzipEnabled := c.gzipEnabled
healthcheckEnabled := c.healthcheckEnabled
retrier := c.retrier
if opt.Retrier != nil {
retrier = opt.Retrier
}
retryStatusCodes := c.retryStatusCodes
if opt.RetryStatusCodes != nil {
retryStatusCodes = opt.RetryStatusCodes
}
defaultHeaders := c.headers
c.mu.RUnlock()
// retry returns true if statusCode indicates the request is to be retried
retry := func(statusCode int) bool {
for _, code := range retryStatusCodes {
if code == statusCode {
return true
}
}
return false
}
var err error
var conn *conn
var req *Request
var resp *Response
var retried bool
var n int
// Change method if sendGetBodyAs is specified.
if opt.Method == "GET" && opt.Body != nil && sendGetBodyAs != "GET" {
opt.Method = sendGetBodyAs
}
for {
pathWithParams := opt.Path
if len(opt.Params) > 0 {
pathWithParams += "?" + opt.Params.Encode()
}
// Get a connection
conn, err = c.next()
if errors.Cause(err) == ErrNoClient {
n++
if !retried {
// Force a healtcheck as all connections seem to be dead.
c.healthcheck(ctx, timeout, false)
if healthcheckEnabled {
retried = true
continue
}
}
wait, ok, rerr := retrier.Retry(ctx, n, nil, nil, err)
if rerr != nil {
return nil, rerr
}
if !ok {
return nil, err
}
retried = true
time.Sleep(wait)
continue // try again
}
if err != nil {
c.errorf("elastic: cannot get connection from pool")
return nil, err
}
req, err = NewRequest(opt.Method, conn.URL()+pathWithParams)
if err != nil {
c.errorf("elastic: cannot create request for %s %s: %v", strings.ToUpper(opt.Method), conn.URL()+pathWithParams, err)
return nil, err
}
if basicAuth {
req.SetBasicAuth(basicAuthUsername, basicAuthPassword)
}
if opt.ContentType != "" {
req.Header.Set("Content-Type", opt.ContentType)
}
for key, value := range opt.Headers {
for _, v := range value {
req.Header.Add(key, v)
}
}
if len(defaultHeaders) > 0 {
for key, value := range defaultHeaders {
for _, v := range value {
req.Header.Add(key, v)
}
}
}
if req.Header.Get("User-Agent") == "" {
req.Header.Set("User-Agent", "elastic/"+Version+" ("+runtime.GOOS+"-"+runtime.GOARCH+")")
}
// Set body
if opt.Body != nil {
err = req.SetBody(opt.Body, gzipEnabled)
if err != nil {
c.errorf("elastic: couldn't set body %+v for request: %v", opt.Body, err)
return nil, err
}
}
// Tracing
c.dumpRequest((*http.Request)(req))
// Get response
res, err := c.c.Do((*http.Request)(req).WithContext(ctx))
if IsContextErr(err) {
// Proceed, but don't mark the node as dead
return nil, err
}
if err != nil {
n++
wait, ok, rerr := retrier.Retry(ctx, n, (*http.Request)(req), res, err)
if rerr != nil {
c.errorf("elastic: %s is dead", conn.URL())
conn.MarkAsDead()
return nil, rerr
}
if !ok {
c.errorf("elastic: %s is dead", conn.URL())
conn.MarkAsDead()
return nil, err
}
retried = true
time.Sleep(wait)
continue // try again
}
if retry(res.StatusCode) {
n++
wait, ok, rerr := retrier.Retry(ctx, n, (*http.Request)(req), res, err)
if rerr != nil {
c.errorf("elastic: %s is dead", conn.URL())
conn.MarkAsDead()
return nil, rerr
}
if ok {
// retry
retried = true
time.Sleep(wait)
continue // try again
}
}
if !opt.Stream {
defer res.Body.Close()
}
// Tracing
c.dumpResponse(res)
// Log deprecation warnings as errors
if len(res.Header["Warning"]) > 0 {
c.deprecationlog((*http.Request)(req), res)
for _, warning := range res.Header["Warning"] {
c.errorf("Deprecation warning: %s", warning)
}
}
// Check for errors
if err := checkResponse((*http.Request)(req), res, opt.IgnoreErrors...); err != nil {
// No retry if request succeeded
// We still try to return a response.
resp, _ = c.newResponse(res, opt.MaxResponseSize, opt.Stream)
return resp, err
}
// We successfully made a request with this connection
conn.MarkAsHealthy()
resp, err = c.newResponse(res, opt.MaxResponseSize, opt.Stream)
if err != nil {
return nil, err
}
break
}
duration := time.Now().UTC().Sub(start)
c.infof("%s %s [status:%d, request:%.3fs]",
strings.ToUpper(opt.Method),
req.URL,
resp.StatusCode,
float64(int64(duration/time.Millisecond))/1000)
return resp, nil
}
// -- Document APIs --
// Index a document.
func (c *Client) Index() *IndexService {
return NewIndexService(c)
}
// Get a document.
func (c *Client) Get() *GetService {
return NewGetService(c)
}
// MultiGet retrieves multiple documents in one roundtrip.
func (c *Client) MultiGet() *MgetService {
return NewMgetService(c)
}
// Mget retrieves multiple documents in one roundtrip.
func (c *Client) Mget() *MgetService {
return NewMgetService(c)
}
// Delete a document.
func (c *Client) Delete() *DeleteService {
return NewDeleteService(c)
}
// DeleteByQuery deletes documents as found by a query.
func (c *Client) DeleteByQuery(indices ...string) *DeleteByQueryService {
return NewDeleteByQueryService(c).Index(indices...)
}
// Update a document.
func (c *Client) Update() *UpdateService {
return NewUpdateService(c)
}
// UpdateByQuery performs an update on a set of documents.
func (c *Client) UpdateByQuery(indices ...string) *UpdateByQueryService {
return NewUpdateByQueryService(c).Index(indices...)
}
// Bulk is the entry point to mass insert/update/delete documents.
func (c *Client) Bulk() *BulkService {
return NewBulkService(c)
}
// BulkProcessor allows setting up a concurrent processor of bulk requests.
func (c *Client) BulkProcessor() *BulkProcessorService {
return NewBulkProcessorService(c)
}
// Reindex copies data from a source index into a destination index.
//
// See https://www.elastic.co/guide/en/elasticsearch/reference/7.0/docs-reindex.html
// for details on the Reindex API.
func (c *Client) Reindex() *ReindexService {
return NewReindexService(c)
}
// TermVectors returns information and statistics on terms in the fields
// of a particular document.
func (c *Client) TermVectors(index string) *TermvectorsService {
builder := NewTermvectorsService(c)
builder = builder.Index(index)
return builder
}
// MultiTermVectors returns information and statistics on terms in the fields
// of multiple documents.
func (c *Client) MultiTermVectors() *MultiTermvectorService {
return NewMultiTermvectorService(c)
}
// -- Search APIs --
// Search is the entry point for searches.
func (c *Client) Search(indices ...string) *SearchService {
return NewSearchService(c).Index(indices...)
}
// MultiSearch is the entry point for multi searches.
func (c *Client) MultiSearch() *MultiSearchService {
return NewMultiSearchService(c)
}
// Count documents.
func (c *Client) Count(indices ...string) *CountService {
return NewCountService(c).Index(indices...)
}
// Explain computes a score explanation for a query and a specific document.
func (c *Client) Explain(index, typ, id string) *ExplainService {
return NewExplainService(c).Index(index).Type(typ).Id(id)
}
// TODO Search Template
// TODO Search Exists API
// Validate allows a user to validate a potentially expensive query without executing it.
func (c *Client) Validate(indices ...string) *ValidateService {
return NewValidateService(c).Index(indices...)
}
// SearchShards returns statistical information about nodes and shards.
func (c *Client) SearchShards(indices ...string) *SearchShardsService {
return NewSearchShardsService(c).Index(indices...)
}
// FieldCaps returns statistical information about fields in indices.
func (c *Client) FieldCaps(indices ...string) *FieldCapsService {
return NewFieldCapsService(c).Index(indices...)
}
// Exists checks if a document exists.
func (c *Client) Exists() *ExistsService {
return NewExistsService(c)
}
// Scroll through documents. Use this to efficiently scroll through results
// while returning the results to a client.
func (c *Client) Scroll(indices ...string) *ScrollService {
return NewScrollService(c).Index(indices...)
}
// ClearScroll can be used to clear search contexts manually.
func (c *Client) ClearScroll(scrollIds ...string) *ClearScrollService {
return NewClearScrollService(c).ScrollId(scrollIds...)
}
// OpenPointInTime opens a new Point in Time.
func (c *Client) OpenPointInTime(indices ...string) *OpenPointInTimeService {
return NewOpenPointInTimeService(c).Index(indices...)
}
// ClosePointInTime closes an existing Point in Time.
func (c *Client) ClosePointInTime(id string) *ClosePointInTimeService {
return NewClosePointInTimeService(c).ID(id)
}
// -- Indices APIs --
// CreateIndex returns a service to create a new index.
func (c *Client) CreateIndex(name string) *IndicesCreateService {
return NewIndicesCreateService(c).Index(name)
}
// DeleteIndex returns a service to delete an index.
func (c *Client) DeleteIndex(indices ...string) *IndicesDeleteService {
return NewIndicesDeleteService(c).Index(indices)
}
// IndexExists allows to check if an index exists.
func (c *Client) IndexExists(indices ...string) *IndicesExistsService {
return NewIndicesExistsService(c).Index(indices)
}
// ShrinkIndex returns a service to shrink one index into another.
func (c *Client) ShrinkIndex(source, target string) *IndicesShrinkService {
return NewIndicesShrinkService(c).Source(source).Target(target)
}
// RolloverIndex rolls an alias over to a new index when the existing index
// is considered to be too large or too old.
func (c *Client) RolloverIndex(alias string) *IndicesRolloverService {
return NewIndicesRolloverService(c).Alias(alias)
}
// IndexStats provides statistics on different operations happining
// in one or more indices.
func (c *Client) IndexStats(indices ...string) *IndicesStatsService {
return NewIndicesStatsService(c).Index(indices...)
}
// OpenIndex opens an index.
func (c *Client) OpenIndex(name string) *IndicesOpenService {
return NewIndicesOpenService(c).Index(name)
}
// CloseIndex closes an index.
func (c *Client) CloseIndex(name string) *IndicesCloseService {
return NewIndicesCloseService(c).Index(name)
}
// FreezeIndex freezes an index.
func (c *Client) FreezeIndex(name string) *IndicesFreezeService {
return NewIndicesFreezeService(c).Index(name)
}
// UnfreezeIndex unfreezes an index.
func (c *Client) UnfreezeIndex(name string) *IndicesUnfreezeService {
return NewIndicesUnfreezeService(c).Index(name)
}
// IndexGet retrieves information about one or more indices.
// IndexGet is only available for Elasticsearch 1.4 or later.
func (c *Client) IndexGet(indices ...string) *IndicesGetService {
return NewIndicesGetService(c).Index(indices...)
}
// IndexGetSettings retrieves settings of all, one or more indices.
func (c *Client) IndexGetSettings(indices ...string) *IndicesGetSettingsService {
return NewIndicesGetSettingsService(c).Index(indices...)
}
// IndexPutSettings sets settings for all, one or more indices.
func (c *Client) IndexPutSettings(indices ...string) *IndicesPutSettingsService {
return NewIndicesPutSettingsService(c).Index(indices...)
}
// IndexSegments retrieves low level segment information for all, one or more indices.
func (c *Client) IndexSegments(indices ...string) *IndicesSegmentsService {
return NewIndicesSegmentsService(c).Index(indices...)
}
// IndexAnalyze performs the analysis process on a text and returns the
// token breakdown of the text.
func (c *Client) IndexAnalyze() *IndicesAnalyzeService {
return NewIndicesAnalyzeService(c)
}
// Forcemerge optimizes one or more indices.
// It replaces the deprecated Optimize API.
func (c *Client) Forcemerge(indices ...string) *IndicesForcemergeService {
return NewIndicesForcemergeService(c).Index(indices...)
}
// Refresh asks Elasticsearch to refresh one or more indices.
func (c *Client) Refresh(indices ...string) *RefreshService {
return NewRefreshService(c).Index(indices...)
}
// Flush asks Elasticsearch to free memory from the index and
// flush data to disk.
func (c *Client) Flush(indices ...string) *IndicesFlushService {
return NewIndicesFlushService(c).Index(indices...)
}
// SyncedFlush performs a synced flush.
//
// See https://www.elastic.co/guide/en/elasticsearch/reference/7.0/indices-synced-flush.html
// for more details on synched flushes and how they differ from a normal
// Flush.
func (c *Client) SyncedFlush(indices ...string) *IndicesSyncedFlushService {
return NewIndicesSyncedFlushService(c).Index(indices...)
}
// ClearCache clears caches for one or more indices.
func (c *Client) ClearCache(indices ...string) *IndicesClearCacheService {
return NewIndicesClearCacheService(c).Index(indices...)
}
// Alias enables the caller to add and/or remove aliases.
func (c *Client) Alias() *AliasService {
return NewAliasService(c)
}
// Aliases returns aliases by index name(s).
func (c *Client) Aliases() *AliasesService {
return NewAliasesService(c)
}
// -- Legacy templates --
// IndexGetTemplate gets an index template (v1/legacy version before 7.8).
//
// This service implements the legacy version of index templates as described
// in https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-templates-v1.html.
//
// See e.g. IndexPutIndexTemplate and IndexPutComponentTemplate for the new version(s).
func (c *Client) IndexGetTemplate(names ...string) *IndicesGetTemplateService {
return NewIndicesGetTemplateService(c).Name(names...)
}
// IndexTemplateExists gets check if an index template exists (v1/legacy version before 7.8).
//
// This service implements the legacy version of index templates as described
// in https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-templates-v1.html.
//
// See e.g. IndexPutIndexTemplate and IndexPutComponentTemplate for the new version(s).
func (c *Client) IndexTemplateExists(name string) *IndicesExistsTemplateService {
return NewIndicesExistsTemplateService(c).Name(name)
}
// IndexPutTemplate creates or updates an index template (v1/legacy version before 7.8).
//
// This service implements the legacy version of index templates as described
// in https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-templates-v1.html.
//
// See e.g. IndexPutIndexTemplate and IndexPutComponentTemplate for the new version(s).
func (c *Client) IndexPutTemplate(name string) *IndicesPutTemplateService {
return NewIndicesPutTemplateService(c).Name(name)
}
// IndexDeleteTemplate deletes an index template (v1/legacy version before 7.8).
//
// This service implements the legacy version of index templates as described
// in https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-templates-v1.html.
//
// See e.g. IndexPutIndexTemplate and IndexPutComponentTemplate for the new version(s).
func (c *Client) IndexDeleteTemplate(name string) *IndicesDeleteTemplateService {
return NewIndicesDeleteTemplateService(c).Name(name)
}
// -- Index templates --
// IndexPutIndexTemplate creates or updates an index template (new version after 7.8).
//
// This service implements the new version of index templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-put-template.html.
//
// See e.g. IndexPutTemplate for the v1/legacy version.
func (c *Client) IndexPutIndexTemplate(name string) *IndicesPutIndexTemplateService {
return NewIndicesPutIndexTemplateService(c).Name(name)
}
// IndexGetIndexTemplate returns an index template (new version after 7.8).
//
// This service implements the new version of index templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-get-template.html.
//
// See e.g. IndexPutTemplate for the v1/legacy version.
func (c *Client) IndexGetIndexTemplate(name string) *IndicesGetIndexTemplateService {
return NewIndicesGetIndexTemplateService(c).Name(name)
}
// IndexDeleteIndexTemplate deletes an index template (new version after 7.8).
//
// This service implements the new version of index templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.9/indices-delete-template.html.
//
// See e.g. IndexPutTemplate for the v1/legacy version.
func (c *Client) IndexDeleteIndexTemplate(name string) *IndicesDeleteIndexTemplateService {
return NewIndicesDeleteIndexTemplateService(c).Name(name)
}
// -- Component templates --
// IndexPutComponentTemplate creates or updates a component template (available since 7.8).
//
// This service implements the component templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.10/indices-component-template.html.
func (c *Client) IndexPutComponentTemplate(name string) *IndicesPutComponentTemplateService {
return NewIndicesPutComponentTemplateService(c).Name(name)
}
// IndexGetComponentTemplate returns a component template (available since 7.8).
//
// This service implements the component templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.10/getting-component-templates.html.
func (c *Client) IndexGetComponentTemplate(name string) *IndicesGetComponentTemplateService {
return NewIndicesGetComponentTemplateService(c).Name(name)
}
// IndexDeleteComponentTemplate deletes a component template (available since 7.8).
//
// This service implements the component templates as described
// on https://www.elastic.co/guide/en/elasticsearch/reference/7.10/indices-delete-component-template.html.
func (c *Client) IndexDeleteComponentTemplate(name string) *IndicesDeleteComponentTemplateService {
return NewIndicesDeleteComponentTemplateService(c).Name(name)
}
// GetMapping gets a mapping.
func (c *Client) GetMapping() *IndicesGetMappingService {
return NewIndicesGetMappingService(c)
}
// PutMapping registers a mapping.
func (c *Client) PutMapping() *IndicesPutMappingService {
return NewIndicesPutMappingService(c)
}
// GetFieldMapping gets mapping for fields.
func (c *Client) GetFieldMapping() *IndicesGetFieldMappingService {
return NewIndicesGetFieldMappingService(c)
}
// -- cat APIs --
// TODO cat master
// TODO cat nodes
// TODO cat pending tasks
// TODO cat plugins
// TODO cat recovery
// TODO cat thread pool
// TODO cat shards
// TODO cat segments
// CatFielddata returns information about the amount of heap memory currently used by the field data cache.
func (c *Client) CatFielddata() *CatFielddataService {
return NewCatFielddataService(c)
}
// CatAliases returns information about aliases.
func (c *Client) CatAliases() *CatAliasesService {
return NewCatAliasesService(c)
}
// CatAllocation returns information about the allocation across nodes.
func (c *Client) CatAllocation() *CatAllocationService {
return NewCatAllocationService(c)
}
// CatCount returns document counts for indices.
func (c *Client) CatCount() *CatCountService {
return NewCatCountService(c)
}
// CatHealth returns information about cluster health.
func (c *Client) CatHealth() *CatHealthService {
return NewCatHealthService(c)
}
// CatIndices returns information about indices.
func (c *Client) CatIndices() *CatIndicesService {
return NewCatIndicesService(c)
}
// CatShards returns information about shards.
func (c *Client) CatShards() *CatShardsService {
return NewCatShardsService(c)
}
// -- Ingest APIs --
// IngestPutPipeline adds pipelines and updates existing pipelines in
// the cluster.
func (c *Client) IngestPutPipeline(id string) *IngestPutPipelineService {
return NewIngestPutPipelineService(c).Id(id)
}
// IngestGetPipeline returns pipelines based on ID.
func (c *Client) IngestGetPipeline(ids ...string) *IngestGetPipelineService {
return NewIngestGetPipelineService(c).Id(ids...)
}
// IngestDeletePipeline deletes a pipeline by ID.
func (c *Client) IngestDeletePipeline(id string) *IngestDeletePipelineService {
return NewIngestDeletePipelineService(c).Id(id)
}
// IngestSimulatePipeline executes a specific pipeline against the set of
// documents provided in the body of the request.
func (c *Client) IngestSimulatePipeline() *IngestSimulatePipelineService {
return NewIngestSimulatePipelineService(c)
}
// -- Cluster APIs --
// ClusterHealth retrieves the health of the cluster.
func (c *Client) ClusterHealth() *ClusterHealthService {
return NewClusterHealthService(c)
}
// ClusterReroute allows for manual changes to the allocation of
// individual shards in the cluster.
func (c *Client) ClusterReroute() *ClusterRerouteService {
return NewClusterRerouteService(c)
}
// ClusterState retrieves the state of the cluster.
func (c *Client) ClusterState() *ClusterStateService {
return NewClusterStateService(c)
}
// ClusterStats retrieves cluster statistics.
func (c *Client) ClusterStats() *ClusterStatsService {
return NewClusterStatsService(c)
}
// NodesInfo retrieves one or more or all of the cluster nodes information.
func (c *Client) NodesInfo() *NodesInfoService {
return NewNodesInfoService(c)
}
// NodesStats retrieves one or more or all of the cluster nodes statistics.
func (c *Client) NodesStats() *NodesStatsService {
return NewNodesStatsService(c)
}
// TasksCancel cancels tasks running on the specified nodes.
func (c *Client) TasksCancel() *TasksCancelService {
return NewTasksCancelService(c)
}
// TasksList retrieves the list of tasks running on the specified nodes.
func (c *Client) TasksList() *TasksListService {
return NewTasksListService(c)
}
// TasksGetTask retrieves a task running on the cluster.
func (c *Client) TasksGetTask() *TasksGetTaskService {
return NewTasksGetTaskService(c)
}
// TODO Pending cluster tasks
// TODO Cluster Reroute
// TODO Cluster Update Settings
// TODO Nodes Stats
// TODO Nodes hot_threads
// -- Snapshot and Restore --
// SnapshotStatus returns information about the status of a snapshot.
func (c *Client) SnapshotStatus() *SnapshotStatusService {
return NewSnapshotStatusService(c)
}
// SnapshotCreate creates a snapshot.
func (c *Client) SnapshotCreate(repository string, snapshot string) *SnapshotCreateService {
return NewSnapshotCreateService(c).Repository(repository).Snapshot(snapshot)
}
// SnapshotCreateRepository creates or updates a snapshot repository.
func (c *Client) SnapshotCreateRepository(repository string) *SnapshotCreateRepositoryService {
return NewSnapshotCreateRepositoryService(c).Repository(repository)
}
// SnapshotDelete deletes a snapshot in a snapshot repository.
func (c *Client) SnapshotDelete(repository string, snapshot string) *SnapshotDeleteService {
return NewSnapshotDeleteService(c).Repository(repository).Snapshot(snapshot)
}
// SnapshotDeleteRepository deletes a snapshot repository.
func (c *Client) SnapshotDeleteRepository(repositories ...string) *SnapshotDeleteRepositoryService {
return NewSnapshotDeleteRepositoryService(c).Repository(repositories...)
}
// SnapshotGetRepository gets a snapshot repository.
func (c *Client) SnapshotGetRepository(repositories ...string) *SnapshotGetRepositoryService {
return NewSnapshotGetRepositoryService(c).Repository(repositories...)
}
// SnapshotGet lists snapshot for a repository.
func (c *Client) SnapshotGet(repository string) *SnapshotGetService {
return NewSnapshotGetService(c).Repository(repository)
}
// SnapshotVerifyRepository verifies a snapshot repository.
func (c *Client) SnapshotVerifyRepository(repository string) *SnapshotVerifyRepositoryService {
return NewSnapshotVerifyRepositoryService(c).Repository(repository)
}
// SnapshotRestore restores the specified indices from a given snapshot
func (c *Client) SnapshotRestore(repository string, snapshot string) *SnapshotRestoreService {
return NewSnapshotRestoreService(c).Repository(repository).Snapshot(snapshot)
}
// -- Scripting APIs --
// GetScript reads a stored script in Elasticsearch.
// Use PutScript for storing a script.
func (c *Client) GetScript() *GetScriptService {
return NewGetScriptService(c)
}
// PutScript allows saving a stored script in Elasticsearch.
func (c *Client) PutScript() *PutScriptService {
return NewPutScriptService(c)
}
// DeleteScript allows removing a stored script from Elasticsearch.
func (c *Client) DeleteScript() *DeleteScriptService {
return NewDeleteScriptService(c)
}
// -- X-Pack General --
// XPackInfo gets information on the xpack plugins enabled on the cluster
func (c *Client) XPackInfo() *XPackInfoService {
return NewXPackInfoService(c)
}
// -- X-Pack Async Search --
// XPackAsyncSearchSubmit starts an asynchronous search.
func (c *Client) XPackAsyncSearchSubmit() *XPackAsyncSearchSubmit {
return NewXPackAsyncSearchSubmit(c)
}
// XPackAsyncSearchGet retrieves the outcome of an asynchronous search.
func (c *Client) XPackAsyncSearchGet() *XPackAsyncSearchGet {
return NewXPackAsyncSearchGet(c)
}
// XPackAsyncSearchDelete deletes an asynchronous search.
func (c *Client) XPackAsyncSearchDelete() *XPackAsyncSearchDelete {
return NewXPackAsyncSearchDelete(c)
}
// -- X-Pack Index Lifecycle Management --
// XPackIlmPutLifecycle adds or modifies an ilm policy.
func (c *Client) XPackIlmPutLifecycle() *XPackIlmPutLifecycleService {
return NewXPackIlmPutLifecycleService(c)
}
// XPackIlmGettLifecycle gets an ilm policy.
func (c *Client) XPackIlmGetLifecycle() *XPackIlmGetLifecycleService {
return NewXPackIlmGetLifecycleService(c)
}
// XPackIlmDeleteLifecycle deletes an ilm policy.
func (c *Client) XPackIlmDeleteLifecycle() *XPackIlmDeleteLifecycleService {
return NewXPackIlmDeleteLifecycleService(c)
}
// -- X-Pack Security --
// XPackSecurityGetRoleMapping gets a role mapping.
func (c *Client) XPackSecurityGetRoleMapping(roleMappingName string) *XPackSecurityGetRoleMappingService {
return NewXPackSecurityGetRoleMappingService(c).Name(roleMappingName)
}
// XPackSecurityPutRoleMapping adds a role mapping.
func (c *Client) XPackSecurityPutRoleMapping(roleMappingName string) *XPackSecurityPutRoleMappingService {
return NewXPackSecurityPutRoleMappingService(c).Name(roleMappingName)
}
// XPackSecurityDeleteRoleMapping deletes a role mapping.
func (c *Client) XPackSecurityDeleteRoleMapping(roleMappingName string) *XPackSecurityDeleteRoleMappingService {
return NewXPackSecurityDeleteRoleMappingService(c).Name(roleMappingName)
}
// XPackSecurityGetRole gets a role.
func (c *Client) XPackSecurityGetRole(roleName string) *XPackSecurityGetRoleService {
return NewXPackSecurityGetRoleService(c).Name(roleName)
}
// XPackSecurityPutRole adds a role.
func (c *Client) XPackSecurityPutRole(roleName string) *XPackSecurityPutRoleService {
return NewXPackSecurityPutRoleService(c).Name(roleName)
}
// XPackSecurityDeleteRole deletes a role.
func (c *Client) XPackSecurityDeleteRole(roleName string) *XPackSecurityDeleteRoleService {
return NewXPackSecurityDeleteRoleService(c).Name(roleName)
}
// TODO: Clear role cache API
// https://www.elastic.co/guide/en/elasticsearch/reference/7.0/security-api-clear-role-cache.html
// XPackSecurityChangePassword changes the password of users in the native realm.
func (c *Client) XPackSecurityChangePassword(username string) *XPackSecurityChangePasswordService {
return NewXPackSecurityChangePasswordService(c).Username(username)
}
// XPackSecurityGetUser gets details about one or more users.
func (c *Client) XPackSecurityGetUser(usernames ...string) *XPackSecurityGetUserService {
return NewXPackSecurityGetUserService(c).Usernames(usernames...)
}
// XPackSecurityPutUser adds or updates a user.
func (c *Client) XPackSecurityPutUser(username string) *XPackSecurityPutUserService {
return NewXPackSecurityPutUserService(c).Username(username)
}
// XPackSecurityEnableUser enables a user.
func (c *Client) XPackSecurityEnableUser(username string) *XPackSecurityEnableUserService {
return NewXPackSecurityEnableUserService(c).Username(username)
}
// XPackSecurityDisableUser disables a user.
func (c *Client) XPackSecurityDisableUser(username string) *XPackSecurityDisableUserService {
return NewXPackSecurityDisableUserService(c).Username(username)
}
// XPackSecurityDeleteUser deletes a user.
func (c *Client) XPackSecurityDeleteUser(username string) *XPackSecurityDeleteUserService {
return NewXPackSecurityDeleteUserService(c).Username(username)
}
// -- X-Pack Watcher --
// XPackWatchPut adds a watch.
func (c *Client) XPackWatchPut(watchId string) *XPackWatcherPutWatchService {
return NewXPackWatcherPutWatchService(c).Id(watchId)
}
// XPackWatchGet gets a watch.
func (c *Client) XPackWatchGet(watchId string) *XPackWatcherGetWatchService {
return NewXPackWatcherGetWatchService(c).Id(watchId)
}
// XPackWatchDelete deletes a watch.
func (c *Client) XPackWatchDelete(watchId string) *XPackWatcherDeleteWatchService {
return NewXPackWatcherDeleteWatchService(c).Id(watchId)
}
// XPackWatchExecute executes a watch.
func (c *Client) XPackWatchExecute() *XPackWatcherExecuteWatchService {
return NewXPackWatcherExecuteWatchService(c)
}
// XPackWatchAck acknowledging a watch.
func (c *Client) XPackWatchAck(watchId string) *XPackWatcherAckWatchService {
return NewXPackWatcherAckWatchService(c).WatchId(watchId)
}
// XPackWatchActivate activates a watch.
func (c *Client) XPackWatchActivate(watchId string) *XPackWatcherActivateWatchService {
return NewXPackWatcherActivateWatchService(c).WatchId(watchId)
}
// XPackWatchDeactivate deactivates a watch.
func (c *Client) XPackWatchDeactivate(watchId string) *XPackWatcherDeactivateWatchService {
return NewXPackWatcherDeactivateWatchService(c).WatchId(watchId)
}
// XPackWatchStats returns the current Watcher metrics.
func (c *Client) XPackWatchStats() *XPackWatcherStatsService {
return NewXPackWatcherStatsService(c)
}
// XPackWatchStart starts a watch.
func (c *Client) XPackWatchStart() *XPackWatcherStartService {
return NewXPackWatcherStartService(c)
}
// XPackWatchStop stops a watch.
func (c *Client) XPackWatchStop() *XPackWatcherStopService {
return NewXPackWatcherStopService(c)
}
// -- Helpers and shortcuts --
// ElasticsearchVersion returns the version number of Elasticsearch
// running on the given URL.
func (c *Client) ElasticsearchVersion(url string) (string, error) {
res, _, err := c.Ping(url).Do(context.Background())
if err != nil {
return "", err
}
return res.Version.Number, nil
}
// IndexNames returns the names of all indices in the cluster.
func (c *Client) IndexNames() ([]string, error) {
res, err := c.IndexGetSettings().Index("_all").Do(context.Background())
if err != nil {
return nil, err
}
var names []string
for name := range res {
names = append(names, name)
}
return names, nil
}
// Ping checks if a given node in a cluster exists and (optionally)
// returns some basic information about the Elasticsearch server,
// e.g. the Elasticsearch version number.
//
// Notice that you need to specify a URL here explicitly.
func (c *Client) Ping(url string) *PingService {
return NewPingService(c).URL(url)
}
// WaitForStatus waits for the cluster to have the given status.
// This is a shortcut method for the ClusterHealth service.
//
// WaitForStatus waits for the specified timeout, e.g. "10s".
// If the cluster will have the given state within the timeout, nil is returned.
// If the request timed out, ErrTimeout is returned.
func (c *Client) WaitForStatus(status string, timeout string) error {
health, err := c.ClusterHealth().WaitForStatus(status).Timeout(timeout).Do(context.Background())
if err != nil {
return err
}
if health.TimedOut {
return ErrTimeout
}
return nil
}
// WaitForGreenStatus waits for the cluster to have the "green" status.
// See WaitForStatus for more details.
func (c *Client) WaitForGreenStatus(timeout string) error {
return c.WaitForStatus("green", timeout)
}
// WaitForYellowStatus waits for the cluster to have the "yellow" status.
// See WaitForStatus for more details.
func (c *Client) WaitForYellowStatus(timeout string) error {
return c.WaitForStatus("yellow", timeout)
}