-
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
package couchbase
import (
"bufio"
"bytes"
"crypto/tls"
"crypto/x509"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"math/rand"
"net/http"
"net/url"
"runtime"
"sort"
"strconv"
"strings"
"sync"
"time"
"unsafe"
"github.com/couchbase/goutils/logging"
"github.com/couchbase/gomemcached" // package name is 'gomemcached'
"github.com/couchbase/gomemcached/client" // package name is 'memcached'
)
// HTTPClient to use for REST and view operations.
var MaxIdleConnsPerHost = 256
var ClientTimeOut = 10 * time.Second
var HTTPTransport = &http.Transport{MaxIdleConnsPerHost: MaxIdleConnsPerHost}
var HTTPClient = &http.Client{Transport: HTTPTransport, Timeout: ClientTimeOut}
// Use this client for reading from streams that should be open for an extended duration.
var HTTPClientForStreaming = &http.Client{Transport: HTTPTransport, Timeout: 0}
// PoolSize is the size of each connection pool (per host).
var PoolSize = 64
// PoolOverflow is the number of overflow connections allowed in a
// pool.
var PoolOverflow = 16
// AsynchronousCloser turns on asynchronous closing for overflow connections
var AsynchronousCloser = false
// TCP KeepAlive enabled/disabled
var TCPKeepalive = false
// Enable MutationToken
var EnableMutationToken = false
// Enable Data Type response
var EnableDataType = false
// Enable Xattr
var EnableXattr = false
// Enable Collections
var EnableCollections = false
// TCP keepalive interval in seconds. Default 30 minutes
var TCPKeepaliveInterval = 30 * 60
// Used to decide whether to skip verification of certificates when
// connecting to an ssl port.
var skipVerify = true
var certFile = ""
var keyFile = ""
var rootFile = ""
func SetSkipVerify(skip bool) {
skipVerify = skip
}
func SetCertFile(cert string) {
certFile = cert
}
func SetKeyFile(cert string) {
keyFile = cert
}
func SetRootFile(cert string) {
rootFile = cert
}
// Allow applications to speciify the Poolsize and Overflow
func SetConnectionPoolParams(size, overflow int) {
if size > 0 {
PoolSize = size
}
if overflow > 0 {
PoolOverflow = overflow
}
}
// Turn off overflow connections
func DisableOverflowConnections() {
PoolOverflow = 0
}
// Toggle asynchronous overflow closer
func EnableAsynchronousCloser(closer bool) {
AsynchronousCloser = closer
}
// Allow TCP keepalive parameters to be set by the application
func SetTcpKeepalive(enabled bool, interval int) {
TCPKeepalive = enabled
if interval > 0 {
TCPKeepaliveInterval = interval
}
}
// AuthHandler is a callback that gets the auth username and password
// for the given bucket.
type AuthHandler interface {
GetCredentials() (string, string, string)
}
// AuthHandler is a callback that gets the auth username and password
// for the given bucket and sasl for memcached.
type AuthWithSaslHandler interface {
AuthHandler
GetSaslCredentials() (string, string)
}
// MultiBucketAuthHandler is kind of AuthHandler that may perform
// different auth for different buckets.
type MultiBucketAuthHandler interface {
AuthHandler
ForBucket(bucket string) AuthHandler
}
// HTTPAuthHandler is kind of AuthHandler that performs more general
// for outgoing http requests than is possible via simple
// GetCredentials() call (i.e. digest auth or different auth per
// different destinations).
type HTTPAuthHandler interface {
AuthHandler
SetCredsForRequest(req *http.Request) error
}
// RestPool represents a single pool returned from the pools REST API.
type RestPool struct {
Name string `json:"name"`
StreamingURI string `json:"streamingUri"`
URI string `json:"uri"`
}
// Pools represents the collection of pools as returned from the REST API.
type Pools struct {
ComponentsVersion map[string]string `json:"componentsVersion,omitempty"`
ImplementationVersion string `json:"implementationVersion"`
IsAdmin bool `json:"isAdminCreds"`
UUID string `json:"uuid"`
Pools []RestPool `json:"pools"`
}
// A Node is a computer in a cluster running the couchbase software.
type Node struct {
ClusterCompatibility int `json:"clusterCompatibility"`
ClusterMembership string `json:"clusterMembership"`
CouchAPIBase string `json:"couchApiBase"`
Hostname string `json:"hostname"`
AlternateNames map[string]NodeAlternateNames `json:"alternateAddresses"`
InterestingStats map[string]float64 `json:"interestingStats,omitempty"`
MCDMemoryAllocated float64 `json:"mcdMemoryAllocated"`
MCDMemoryReserved float64 `json:"mcdMemoryReserved"`
MemoryFree float64 `json:"memoryFree"`
MemoryTotal float64 `json:"memoryTotal"`
OS string `json:"os"`
Ports map[string]int `json:"ports"`
Services []string `json:"services"`
Status string `json:"status"`
Uptime int `json:"uptime,string"`
Version string `json:"version"`
ThisNode bool `json:"thisNode,omitempty"`
}
// A Pool of nodes and buckets.
type Pool struct {
BucketMap map[string]*Bucket
Nodes []Node
BucketURL map[string]string `json:"buckets"`
MemoryQuota float64 `json:"memoryQuota"`
CbasMemoryQuota float64 `json:"cbasMemoryQuota"`
EventingMemoryQuota float64 `json:"eventingMemoryQuota"`
FtsMemoryQuota float64 `json:"ftsMemoryQuota"`
IndexMemoryQuota float64 `json:"indexMemoryQuota"`
client *Client
}
// VBucketServerMap is the a mapping of vbuckets to nodes.
type VBucketServerMap struct {
HashAlgorithm string `json:"hashAlgorithm"`
NumReplicas int `json:"numReplicas"`
ServerList []string `json:"serverList"`
VBucketMap [][]int `json:"vBucketMap"`
}
type DurablitySettings struct {
Persist PersistTo
Observe ObserveTo
}
// Bucket is the primary entry point for most data operations.
// Bucket is a locked data structure. All access to its fields should be done using read or write locking,
// as appropriate.
//
// Some access methods require locking, but rely on the caller to do so. These are appropriate
// for calls from methods that have already locked the structure. Methods like this
// take a boolean parameter "bucketLocked".
type Bucket struct {
sync.RWMutex
AuthType string `json:"authType"`
Capabilities []string `json:"bucketCapabilities"`
CapabilitiesVersion string `json:"bucketCapabilitiesVer"`
CollectionsManifestUid string `json:"collectionsManifestUid"`
Type string `json:"bucketType"`
Name string `json:"name"`
NodeLocator string `json:"nodeLocator"`
Quota map[string]float64 `json:"quota,omitempty"`
Replicas int `json:"replicaNumber"`
Password string `json:"saslPassword"`
URI string `json:"uri"`
StreamingURI string `json:"streamingUri"`
LocalRandomKeyURI string `json:"localRandomKeyUri,omitempty"`
UUID string `json:"uuid"`
ConflictResolutionType string `json:"conflictResolutionType,omitempty"`
DDocs struct {
URI string `json:"uri"`
} `json:"ddocs,omitempty"`
BasicStats map[string]interface{} `json:"basicStats,omitempty"`
Controllers map[string]interface{} `json:"controllers,omitempty"`
// These are used for JSON IO, but isn't used for processing
// since it needs to be swapped out safely.
VBSMJson VBucketServerMap `json:"vBucketServerMap"`
NodesJSON []Node `json:"nodes"`
pool *Pool
connPools unsafe.Pointer // *[]*connectionPool
vBucketServerMap unsafe.Pointer // *VBucketServerMap
nodeList unsafe.Pointer // *[]Node
commonSufix string
ah AuthHandler // auth handler
ds *DurablitySettings // Durablity Settings for this bucket
closed bool
}
// PoolServices is all the bucket-independent services in a pool
type PoolServices struct {
Rev int `json:"rev"`
NodesExt []NodeServices `json:"nodesExt"`
Capabilities json.RawMessage `json:"clusterCapabilities"`
}
// NodeServices is all the bucket-independent services running on
// a node (given by Hostname)
type NodeServices struct {
Services map[string]int `json:"services,omitempty"`
Hostname string `json:"hostname"`
ThisNode bool `json:"thisNode"`
AlternateNames map[string]NodeAlternateNames `json:"alternateAddresses"`
}
type NodeAlternateNames struct {
Hostname string `json:"hostname"`
Ports map[string]int `json:"ports"`
}
type BucketNotFoundError struct {
bucket string
}
func (e *BucketNotFoundError) Error() string {
return fmt.Sprint("No bucket named " + e.bucket)
}
type BucketAuth struct {
name string
saslPwd string
bucket string
}
func newBucketAuth(name string, pass string, bucket string) *BucketAuth {
return &BucketAuth{name: name, saslPwd: pass, bucket: bucket}
}
func (ba *BucketAuth) GetCredentials() (string, string, string) {
return ba.name, ba.saslPwd, ba.bucket
}
// VBServerMap returns the current VBucketServerMap.
func (b *Bucket) VBServerMap() *VBucketServerMap {
b.RLock()
defer b.RUnlock()
ret := (*VBucketServerMap)(b.vBucketServerMap)
return ret
}
func (b *Bucket) GetVBmap(addrs []string) (map[string][]uint16, error) {
vbmap := b.VBServerMap()
servers := vbmap.ServerList
if addrs == nil {
addrs = vbmap.ServerList
}
m := make(map[string][]uint16)
for _, addr := range addrs {
m[addr] = make([]uint16, 0)
}
for vbno, idxs := range vbmap.VBucketMap {
if len(idxs) == 0 {
return nil, fmt.Errorf("vbmap: No KV node no for vb %d", vbno)
} else if idxs[0] < 0 || idxs[0] >= len(servers) {
return nil, fmt.Errorf("vbmap: Invalid KV node no %d for vb %d", idxs[0], vbno)
}
addr := servers[idxs[0]]
if _, ok := m[addr]; ok {
m[addr] = append(m[addr], uint16(vbno))
}
}
return m, nil
}
// true if node is not on the bucket VBmap
func (b *Bucket) checkVBmap(node string) bool {
vbmap := b.VBServerMap()
servers := vbmap.ServerList
for _, idxs := range vbmap.VBucketMap {
if len(idxs) == 0 {
return true
} else if idxs[0] < 0 || idxs[0] >= len(servers) {
return true
}
if servers[idxs[0]] == node {
return false
}
}
return true
}
func (b *Bucket) GetName() string {
b.RLock()
defer b.RUnlock()
ret := b.Name
return ret
}
func (b *Bucket) GetUUID() string {
b.RLock()
defer b.RUnlock()
ret := b.UUID
return ret
}
// Nodes returns the current list of nodes servicing this bucket.
func (b *Bucket) Nodes() []Node {
b.RLock()
defer b.RUnlock()
ret := *(*[]Node)(b.nodeList)
return ret
}
// return the list of healthy nodes
func (b *Bucket) HealthyNodes() []Node {
nodes := []Node{}
for _, n := range b.Nodes() {
if n.Status == "healthy" && n.CouchAPIBase != "" {
nodes = append(nodes, n)
}
if n.Status != "healthy" { // log non-healthy node
logging.Infof("Non-healthy node; node details:")
logging.Infof("Hostname=%v, Status=%v, CouchAPIBase=%v, ThisNode=%v", n.Hostname, n.Status, n.CouchAPIBase, n.ThisNode)
}
}
return nodes
}
func (b *Bucket) getConnPools(bucketLocked bool) []*connectionPool {
if !bucketLocked {
b.RLock()
defer b.RUnlock()
}
if b.connPools != nil {
return *(*[]*connectionPool)(b.connPools)
} else {
return nil
}
}
func (b *Bucket) replaceConnPools(with []*connectionPool) {
b.Lock()
defer b.Unlock()
old := b.connPools
b.connPools = unsafe.Pointer(&with)
if old != nil {
for _, pool := range *(*[]*connectionPool)(old) {
if pool != nil {
pool.Close()
}
}
}
return
}
func (b *Bucket) getConnPool(i int) *connectionPool {
if i < 0 {
return nil
}
p := b.getConnPools(false /* not already locked */)
if len(p) > i {
return p[i]
}
return nil
}
func (b *Bucket) getConnPoolByHost(host string, bucketLocked bool) *connectionPool {
pools := b.getConnPools(bucketLocked)
for _, p := range pools {
if p != nil && p.host == host {
return p
}
}
return nil
}
// Given a vbucket number, returns a memcached connection to it.
// The connection must be returned to its pool after use.
func (b *Bucket) getConnectionToVBucket(vb uint32) (*memcached.Client, *connectionPool, error) {
for {
vbm := b.VBServerMap()
if len(vbm.VBucketMap) < int(vb) {
return nil, nil, fmt.Errorf("go-couchbase: vbmap smaller than vbucket list: %v vs. %v",
vb, vbm.VBucketMap)
}
masterId := vbm.VBucketMap[vb][0]
if masterId < 0 {
return nil, nil, fmt.Errorf("go-couchbase: No master for vbucket %d", vb)
}
pool := b.getConnPool(masterId)
conn, err := pool.Get()
if err != errClosedPool {
return conn, pool, err
}
// If conn pool was closed, because another goroutine refreshed the vbucket map, retry...
}
}
// To get random documents, we need to cover all the nodes, so select
// a connection at random.
func (b *Bucket) getRandomConnection() (*memcached.Client, *connectionPool, error) {
for {
var currentPool = 0
pools := b.getConnPools(false /* not already locked */)
if len(pools) == 0 {
return nil, nil, fmt.Errorf("No connection pool found")
} else if len(pools) > 1 { // choose a random connection
currentPool = rand.Intn(len(pools))
} // if only one pool, currentPool defaults to 0, i.e., the only pool
// get the pool
pool := pools[currentPool]
conn, err := pool.Get()
if err != errClosedPool {
return conn, pool, err
}
// If conn pool was closed, because another goroutine refreshed the vbucket map, retry...
}
}
//
// Get a random document from a bucket. Since the bucket may be distributed
// across nodes, we must first select a random connection, and then use the
// Client.GetRandomDoc() call to get a random document from that node.
//
func (b *Bucket) GetRandomDoc(context ...*memcached.ClientContext) (*gomemcached.MCResponse, error) {
// get a connection from the pool
conn, pool, err := b.getRandomConnection()
if err != nil {
return nil, err
}
conn.SetDeadline(getDeadline(time.Time{}, DefaultTimeout))
// We may need to select the bucket before GetRandomDoc()
// will work. This is sometimes done at startup (see defaultMkConn())
// but not always, depending on the auth type.
if conn.LastBucket() != b.Name {
_, err = conn.SelectBucket(b.Name)
if err != nil {
return nil, err
}
}
// get a randomm document from the connection
doc, err := conn.GetRandomDoc(context...)
// need to return the connection to the pool
pool.Return(conn)
return doc, err
}
// Bucket DDL
func uriAdj(s string) string {
return strings.Replace(s, "%", "%25", -1)
}
func (b *Bucket) CreateScope(scope string) error {
b.RLock()
pool := b.pool
client := pool.client
b.RUnlock()
args := map[string]interface{}{"name": scope}
return client.parsePostURLResponseTerse("/pools/default/buckets/"+uriAdj(b.Name)+"/scopes", args, nil)
}
func (b *Bucket) DropScope(scope string) error {
b.RLock()
pool := b.pool
client := pool.client
b.RUnlock()
return client.parseDeleteURLResponseTerse("/pools/default/buckets/"+uriAdj(b.Name)+"/scopes/"+uriAdj(scope), nil, nil)
}
func (b *Bucket) CreateCollection(scope string, collection string) error {
b.RLock()
pool := b.pool
client := pool.client
b.RUnlock()
args := map[string]interface{}{"name": collection}
return client.parsePostURLResponseTerse("/pools/default/buckets/"+uriAdj(b.Name)+"/scopes/"+uriAdj(scope)+"/collections", args, nil)
}
func (b *Bucket) DropCollection(scope string, collection string) error {
b.RLock()
pool := b.pool
client := pool.client
b.RUnlock()
return client.parseDeleteURLResponseTerse("/pools/default/buckets/"+uriAdj(b.Name)+"/scopes/"+uriAdj(scope)+"/collections/"+uriAdj(collection), nil, nil)
}
func (b *Bucket) FlushCollection(scope string, collection string) error {
b.RLock()
pool := b.pool
client := pool.client
b.RUnlock()
args := map[string]interface{}{"name": collection, "scope": scope}
return client.parsePostURLResponseTerse("/pools/default/buckets/"+uriAdj(b.Name)+"/collections-flush", args, nil)
}
func (b *Bucket) getMasterNode(i int) string {
p := b.getConnPools(false /* not already locked */)
if len(p) > i {
return p[i].host
}
return ""
}
func (b *Bucket) authHandler(bucketLocked bool) (ah AuthHandler) {
if !bucketLocked {
b.RLock()
defer b.RUnlock()
}
pool := b.pool
name := b.Name
if pool != nil {
ah = pool.client.ah
}
if mbah, ok := ah.(MultiBucketAuthHandler); ok {
return mbah.ForBucket(name)
}
if ah == nil {
ah = &basicAuth{name, ""}
}
return
}
// NodeAddresses gets the (sorted) list of memcached node addresses
// (hostname:port).
func (b *Bucket) NodeAddresses() []string {
vsm := b.VBServerMap()
rv := make([]string, len(vsm.ServerList))
copy(rv, vsm.ServerList)
sort.Strings(rv)
return rv
}
// CommonAddressSuffix finds the longest common suffix of all
// host:port strings in the node list.
func (b *Bucket) CommonAddressSuffix() string {
input := []string{}
for _, n := range b.Nodes() {
input = append(input, n.Hostname)
}
return FindCommonSuffix(input)
}
// A Client is the starting point for all services across all buckets
// in a Couchbase cluster.
type Client struct {
BaseURL *url.URL
ah AuthHandler
Info Pools
tlsConfig *tls.Config
}
func maybeAddAuth(req *http.Request, ah AuthHandler) error {
if hah, ok := ah.(HTTPAuthHandler); ok {
return hah.SetCredsForRequest(req)
}
if ah != nil {
user, pass, _ := ah.GetCredentials()
req.Header.Set("Authorization", "Basic "+
base64.StdEncoding.EncodeToString([]byte(user+":"+pass)))
}
return nil
}
// arbitary number, may need to be tuned #FIXME
const HTTP_MAX_RETRY = 5
// Someday golang network packages will implement standard
// error codes. Until then #sigh
func isHttpConnError(err error) bool {
estr := err.Error()
return strings.Contains(estr, "broken pipe") ||
strings.Contains(estr, "broken connection") ||
strings.Contains(estr, "connection reset")
}
var client *http.Client
var clientForStreaming *http.Client
func ClientConfigForX509(certFile, keyFile, rootFile string) (*tls.Config, error) {
cfg := &tls.Config{}
if certFile != "" && keyFile != "" {
tlsCert, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return nil, err
}
cfg.Certificates = []tls.Certificate{tlsCert}
} else {
//error need to pass both certfile and keyfile
return nil, fmt.Errorf("N1QL: Need to pass both certfile and keyfile")
}
var caCert []byte
var err1 error
caCertPool := x509.NewCertPool()
if rootFile != "" {
// Read that value in
caCert, err1 = ioutil.ReadFile(rootFile)
if err1 != nil {
return nil, fmt.Errorf(" Error in reading cacert file, err: %v", err1)
}
caCertPool.AppendCertsFromPEM(caCert)
}
cfg.RootCAs = caCertPool
return cfg, nil
}
// This version of doHTTPRequest is for requests where the response connection is held open
// for an extended duration since line is a new and significant output.
//
// The ordinary version of this method expects the results to arrive promptly, and
// therefore use an HTTP client with a timeout. This client is not suitable
// for streaming use.
func doHTTPRequestForStreaming(req *http.Request) (*http.Response, error) {
var err error
var res *http.Response
// we need a client that ignores certificate errors, since we self-sign
// our certs
if clientForStreaming == nil && req.URL.Scheme == "https" {
var tr *http.Transport
if skipVerify {
tr = &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
MaxIdleConnsPerHost: MaxIdleConnsPerHost,
}
} else {
// Handle cases with cert
cfg, err := ClientConfigForX509(certFile, keyFile, rootFile)
if err != nil {
return nil, err
}
tr = &http.Transport{
TLSClientConfig: cfg,
MaxIdleConnsPerHost: MaxIdleConnsPerHost,
}
}
clientForStreaming = &http.Client{Transport: tr, Timeout: 0}
} else if clientForStreaming == nil {
clientForStreaming = HTTPClientForStreaming
}
for i := 0; i < HTTP_MAX_RETRY; i++ {
res, err = clientForStreaming.Do(req)
if err != nil && isHttpConnError(err) {
continue
}
break
}
if err != nil {
return nil, err
}
return res, err
}
func doHTTPRequest(req *http.Request) (*http.Response, error) {
var err error
var res *http.Response
// we need a client that ignores certificate errors, since we self-sign
// our certs
if client == nil && req.URL.Scheme == "https" {
var tr *http.Transport
if skipVerify {
tr = &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
MaxIdleConnsPerHost: MaxIdleConnsPerHost,
}
} else {
// Handle cases with cert
cfg, err := ClientConfigForX509(certFile, keyFile, rootFile)
if err != nil {
return nil, err
}
tr = &http.Transport{
TLSClientConfig: cfg,
MaxIdleConnsPerHost: MaxIdleConnsPerHost,
}
}
client = &http.Client{Transport: tr, Timeout: ClientTimeOut}
} else if client == nil {
client = HTTPClient
}
for i := 0; i < HTTP_MAX_RETRY; i++ {
res, err = client.Do(req)
if err != nil && isHttpConnError(err) {
continue
}
break
}
if err != nil {
return nil, err
}
return res, err
}
func doPutAPI(baseURL *url.URL, path string, params map[string]interface{}, authHandler AuthHandler, out interface{}, terse bool) error {
return doOutputAPI("PUT", baseURL, path, params, authHandler, out, terse)
}
func doPostAPI(baseURL *url.URL, path string, params map[string]interface{}, authHandler AuthHandler, out interface{}, terse bool) error {
return doOutputAPI("POST", baseURL, path, params, authHandler, out, terse)
}
func doDeleteAPI(baseURL *url.URL, path string, params map[string]interface{}, authHandler AuthHandler, out interface{}, terse bool) error {
return doOutputAPI("DELETE", baseURL, path, params, authHandler, out, terse)
}
func doOutputAPI(
httpVerb string,
baseURL *url.URL,
path string,
params map[string]interface{},
authHandler AuthHandler,
out interface{},
terse bool) error {
var requestUrl string
if q := strings.Index(path, "?"); q > 0 {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path[:q] + "?" + path[q+1:]
} else {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path
}
postData := url.Values{}
for k, v := range params {
postData.Set(k, fmt.Sprintf("%v", v))
}
req, err := http.NewRequest(httpVerb, requestUrl, bytes.NewBufferString(postData.Encode()))
if err != nil {
return err
}
req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
err = maybeAddAuth(req, authHandler)
if err != nil {
return err
}
res, err := doHTTPRequest(req)
if err != nil {
return err
}
defer res.Body.Close()
// 200 - ok, 202 - accepted (asynchronously)
if res.StatusCode != 200 && res.StatusCode != 202 {
bod, _ := ioutil.ReadAll(io.LimitReader(res.Body, 512))
if terse {
var outBuf interface{}
err := json.Unmarshal(bod, &outBuf)
if err == nil && outBuf != nil {
switch errText := outBuf.(type) {
case string:
return fmt.Errorf("%s", errText)
case map[string]interface{}:
errField := errText["errors"]
if errField != nil {
// remove annoying 'map' prefix
return fmt.Errorf("%s", strings.TrimPrefix(fmt.Sprintf("%v", errField), "map"))
}
}
}
return fmt.Errorf("%s", string(bod))
}
return fmt.Errorf("HTTP error %v getting %q: %s",
res.Status, requestUrl, bod)
}
d := json.NewDecoder(res.Body)
// PUT/POST/DELETE request may not have a response body
if d.More() {
if err = d.Decode(&out); err != nil {
return err
}
}
return nil
}
func queryRestAPI(
baseURL *url.URL,
path string,
authHandler AuthHandler,
out interface{},
terse bool) error {
var requestUrl string
if q := strings.Index(path, "?"); q > 0 {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path[:q] + "?" + path[q+1:]
} else {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path
}
req, err := http.NewRequest("GET", requestUrl, nil)
if err != nil {
return err
}
err = maybeAddAuth(req, authHandler)
if err != nil {
return err
}
res, err := doHTTPRequest(req)
if err != nil {
return err
}
defer res.Body.Close()
if res.StatusCode != 200 {
bod, _ := ioutil.ReadAll(io.LimitReader(res.Body, 512))
if terse {
var outBuf interface{}
err := json.Unmarshal(bod, &outBuf)
if err == nil && outBuf != nil {
errText, ok := outBuf.(string)
if ok {
return fmt.Errorf(errText)
}
}
return fmt.Errorf(string(bod))
}
return fmt.Errorf("HTTP error %v getting %q: %s",
res.Status, requestUrl, bod)
}
d := json.NewDecoder(res.Body)
// GET request should have a response body
if err = d.Decode(&out); err != nil {
return fmt.Errorf("json decode err: %#v, for requestUrl: %s",
err, requestUrl)
}
return nil
}
func (c *Client) ProcessStream(path string, callb func(interface{}) error, data interface{}) error {
return c.processStream(c.BaseURL, path, c.ah, callb, data)
}
// Based on code in http://src.couchbase.org/source/xref/trunk/goproj/src/github.com/couchbase/indexing/secondary/dcp/pools.go#309
func (c *Client) processStream(baseURL *url.URL, path string, authHandler AuthHandler, callb func(interface{}) error, data interface{}) error {
var requestUrl string
if q := strings.Index(path, "?"); q > 0 {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path[:q] + "?" + path[q+1:]
} else {
requestUrl = baseURL.Scheme + "://" + baseURL.Host + path
}
req, err := http.NewRequest("GET", requestUrl, nil)
if err != nil {
return err
}
err = maybeAddAuth(req, authHandler)
if err != nil {
return err
}
res, err := doHTTPRequestForStreaming(req)
if err != nil {
return err
}
defer res.Body.Close()
if res.StatusCode != 200 {
bod, _ := ioutil.ReadAll(io.LimitReader(res.Body, 512))
return fmt.Errorf("HTTP error %v getting %q: %s",
res.Status, requestUrl, bod)
}
reader := bufio.NewReader(res.Body)
for {
bs, err := reader.ReadBytes('\n')
if err != nil {
return err
}
if len(bs) == 1 && bs[0] == '\n' {
continue
}
err = json.Unmarshal(bs, data)
if err != nil {
return err
}
err = callb(data)
if err != nil {
return err
}
}
return nil
}
func (c *Client) parseURLResponse(path string, out interface{}) error {
return queryRestAPI(c.BaseURL, path, c.ah, out, false)
}
func (c *Client) parsePostURLResponse(path string, params map[string]interface{}, out interface{}) error {
return doPostAPI(c.BaseURL, path, params, c.ah, out, false)
}
func (c *Client) parsePostURLResponseTerse(path string, params map[string]interface{}, out interface{}) error {
return doPostAPI(c.BaseURL, path, params, c.ah, out, true)
}
func (c *Client) parseDeleteURLResponse(path string, params map[string]interface{}, out interface{}) error {
return doDeleteAPI(c.BaseURL, path, params, c.ah, out, false)
}
func (c *Client) parseDeleteURLResponseTerse(path string, params map[string]interface{}, out interface{}) error {
return doDeleteAPI(c.BaseURL, path, params, c.ah, out, true)
}
func (c *Client) parsePutURLResponse(path string, params map[string]interface{}, out interface{}) error {
return doPutAPI(c.BaseURL, path, params, c.ah, out, false)
}
func (c *Client) parsePutURLResponseTerse(path string, params map[string]interface{}, out interface{}) error {
return doPutAPI(c.BaseURL, path, params, c.ah, out, true)
}
func (b *Bucket) parseURLResponse(path string, out interface{}) error {
nodes := b.Nodes()
if len(nodes) == 0 {
return errors.New("no couch rest URLs")
}
// Pick a random node to start querying.
startNode := rand.Intn(len(nodes))
maxRetries := len(nodes)
for i := 0; i < maxRetries; i++ {
node := nodes[(startNode+i)%len(nodes)] // Wrap around the nodes list.
// Skip non-healthy nodes.
if node.Status != "healthy" || node.CouchAPIBase == "" {
continue
}
url := &url.URL{
Host: node.Hostname,
Scheme: "http",
}
// Lock here to avoid having pool closed under us.
b.RLock()
err := queryRestAPI(url, path, b.pool.client.ah, out, false)
b.RUnlock()
if err == nil {
return err
}
}
return errors.New("All nodes failed to respond or no healthy nodes for bucket found")
}
func (b *Bucket) parseAPIResponse(path string, out interface{}) error {
nodes := b.Nodes()
if len(nodes) == 0 {
return errors.New("no couch rest URLs")
}
var err error
var u *url.URL
// Pick a random node to start querying.
startNode := rand.Intn(len(nodes))
maxRetries := len(nodes)
for i := 0; i < maxRetries; i++ {
node := nodes[(startNode+i)%len(nodes)] // Wrap around the nodes list.
// Skip non-healthy nodes.
if node.Status != "healthy" || node.CouchAPIBase == "" {
continue
}
u, err = ParseURL(node.CouchAPIBase)
// Lock here so pool does not get closed under us.
b.RLock()
if err != nil {
b.RUnlock()
return fmt.Errorf("config error: Bucket %q node #%d CouchAPIBase=%q: %v",
b.Name, i, node.CouchAPIBase, err)
} else if b.pool != nil {
u.User = b.pool.client.BaseURL.User
}
u.Path = path
// generate the path so that the strings are properly escaped
// MB-13770
requestPath := strings.Split(u.String(), u.Host)[1]
err = queryRestAPI(u, requestPath, b.pool.client.ah, out, false)
b.RUnlock()
if err == nil {
return err
}
}
var errStr string
if err != nil {
errStr = "Error " + err.Error()
}
return errors.New("All nodes failed to respond or returned error or no healthy nodes for bucket found." + errStr)
}
type basicAuth struct {
u, p string
}
func (b basicAuth) GetCredentials() (string, string, string) {
return b.u, b.p, b.u
}
func basicAuthFromURL(us string) (ah AuthHandler) {
u, err := ParseURL(us)
if err != nil {
return
}
if user := u.User; user != nil {
pw, _ := user.Password()
ah = basicAuth{user.Username(), pw}
}
return
}
// ConnectWithAuth connects to a couchbase cluster with the given
// authentication handler.
func ConnectWithAuth(baseU string, ah AuthHandler) (c Client, err error) {
c.BaseURL, err = ParseURL(baseU)
if err != nil {
return
}
c.ah = ah
return c, c.parseURLResponse("/pools", &c.Info)
}
// Call this method with a TLS certificate file name to make communication
// with the KV engine encrypted.
//
// This method should be called immediately after a Connect*() method.
func (c *Client) InitTLS(certFile string) error {
serverCert, err := ioutil.ReadFile(certFile)
if err != nil {
return err
}
CA_Pool := x509.NewCertPool()
CA_Pool.AppendCertsFromPEM(serverCert)
c.tlsConfig = &tls.Config{RootCAs: CA_Pool}
return nil
}
func (c *Client) ClearTLS() {
c.tlsConfig = nil
}
// ConnectWithAuthCreds connects to a couchbase cluster with the give
// authorization creds returned by cb_auth
func ConnectWithAuthCreds(baseU, username, password string) (c Client, err error) {
c.BaseURL, err = ParseURL(baseU)
if err != nil {
return
}
c.ah = newBucketAuth(username, password, "")
return c, c.parseURLResponse("/pools", &c.Info)
}
// Connect to a couchbase cluster. An authentication handler will be
// created from the userinfo in the URL if provided.
func Connect(baseU string) (Client, error) {
return ConnectWithAuth(baseU, basicAuthFromURL(baseU))
}
type BucketInfo struct {
Name string // name of bucket
Password string // SASL password of bucket
}
//Get SASL buckets
func GetBucketList(baseU string) (bInfo []BucketInfo, err error) {
c := &Client{}
c.BaseURL, err = ParseURL(baseU)
if err != nil {
return
}
c.ah = basicAuthFromURL(baseU)
var buckets []Bucket
err = c.parseURLResponse("/pools/default/buckets", &buckets)
if err != nil {
return
}
bInfo = make([]BucketInfo, 0)
for _, bucket := range buckets {
bucketInfo := BucketInfo{Name: bucket.Name, Password: bucket.Password}
bInfo = append(bInfo, bucketInfo)
}
return bInfo, err
}
//Set viewUpdateDaemonOptions
func SetViewUpdateParams(baseU string, params map[string]interface{}) (viewOpts map[string]interface{}, err error) {
c := &Client{}
c.BaseURL, err = ParseURL(baseU)
if err != nil {
return
}
c.ah = basicAuthFromURL(baseU)
if len(params) < 1 {
return nil, fmt.Errorf("No params to set")
}
err = c.parsePostURLResponse("/settings/viewUpdateDaemon", params, &viewOpts)
if err != nil {
return
}
return viewOpts, err
}
// This API lets the caller know, if the list of nodes a bucket is
// connected to has gone through an edit (a rebalance operation)
// since the last update to the bucket, in which case a Refresh is
// advised.
func (b *Bucket) NodeListChanged() bool {
b.RLock()
pool := b.pool
uri := b.URI
b.RUnlock()
tmpb := &Bucket{}
err := pool.client.parseURLResponse(uri, tmpb)
if err != nil {
return true
}
bNodes := *(*[]Node)(b.nodeList)
if len(bNodes) != len(tmpb.NodesJSON) {
return true
}
bucketHostnames := map[string]bool{}
for _, node := range bNodes {
bucketHostnames[node.Hostname] = true
}
for _, node := range tmpb.NodesJSON {
if _, found := bucketHostnames[node.Hostname]; !found {
return true
}
}
return false
}
// Sample data for scopes and collections as returned from the
// /pooles/default/$BUCKET_NAME/collections API.
// {"myScope2":{"myCollectionC":{}},"myScope1":{"myCollectionB":{},"myCollectionA":{}},"_default":{"_default":{}}}
// Structures for parsing collections manifest.
// The map key is the name of the scope.
// Example data:
// {"uid":"b","scopes":[
// {"name":"_default","uid":"0","collections":[
// {"name":"_default","uid":"0"}]},
// {"name":"myScope1","uid":"8","collections":[
// {"name":"myCollectionB","uid":"c"},
// {"name":"myCollectionA","uid":"b"}]},
// {"name":"myScope2","uid":"9","collections":[
// {"name":"myCollectionC","uid":"d"}]}]}
type InputManifest struct {
Uid string
Scopes []InputScope
}
type InputScope struct {
Name string
Uid string
Collections []InputCollection
}
type InputCollection struct {
Name string
Uid string
}
// Structures for storing collections information.
type Manifest struct {
Uid uint64
Scopes map[string]*Scope // map by name
}
type Scope struct {
Name string
Uid uint64
Collections map[string]*Collection // map by name
}
type Collection struct {
Name string
Uid uint64
}
var _EMPTY_MANIFEST *Manifest = &Manifest{Uid: 0, Scopes: map[string]*Scope{}}
func parseCollectionsManifest(res *gomemcached.MCResponse) (*Manifest, error) {
if !EnableCollections {
return _EMPTY_MANIFEST, nil
}
var im InputManifest
err := json.Unmarshal(res.Body, &im)
if err != nil {
return nil, err
}
uid, err := strconv.ParseUint(im.Uid, 16, 64)
if err != nil {
return nil, err
}
mani := &Manifest{Uid: uid, Scopes: make(map[string]*Scope, len(im.Scopes))}
for _, iscope := range im.Scopes {
scope_uid, err := strconv.ParseUint(iscope.Uid, 16, 64)
if err != nil {
return nil, err
}
scope := &Scope{Uid: scope_uid, Name: iscope.Name, Collections: make(map[string]*Collection, len(iscope.Collections))}
mani.Scopes[iscope.Name] = scope
for _, icoll := range iscope.Collections {
coll_uid, err := strconv.ParseUint(icoll.Uid, 16, 64)
if err != nil {
return nil, err
}
coll := &Collection{Uid: coll_uid, Name: icoll.Name}
scope.Collections[icoll.Name] = coll
}
}
return mani, nil
}
// This function assumes the bucket is locked.
func (b *Bucket) GetCollectionsManifest() (*Manifest, error) {
// Collections not used?
if !EnableCollections {
return nil, fmt.Errorf("Collections not enabled.")
}
b.RLock()
pools := b.getConnPools(true /* already locked */)
if len(pools) == 0 {
b.RUnlock()
return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest: no connection pool. No collections access to bucket %s.", b.Name)
}
pool := pools[0] // Any pool will do, so use the first one.
b.RUnlock()
client, err := pool.Get()
if err != nil {
return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest: %v. No collections access to bucket %s.", err, b.Name)
}
client.SetDeadline(getDeadline(time.Time{}, DefaultTimeout))
// We need to select the bucket before GetCollectionsManifest()
// will work. This is sometimes done at startup (see defaultMkConn())
// but not always, depending on the auth type.
// Doing this is safe because we collect the the connections
// by bucket, so the bucket being selected will never change.
_, err = client.SelectBucket(b.Name)
if err != nil {
pool.Return(client)
return nil, fmt.Errorf("Unable to select bucket %s: %v. No collections access to bucket %s.", err, b.Name, b.Name)
}
res, err := client.GetCollectionsManifest()
if err != nil {
pool.Return(client)
return nil, fmt.Errorf("Unable to retrieve collections manifest: %v. No collections access to bucket %s.", err, b.Name)
}
mani, err := parseCollectionsManifest(res)
if err != nil {
pool.Return(client)
return nil, fmt.Errorf("Unable to parse collections manifest: %v. No collections access to bucket %s.", err, b.Name)
}
pool.Return(client)
return mani, nil
}
func (b *Bucket) RefreshFully() error {
return b.refresh(false)
}
func (b *Bucket) Refresh() error {
return b.refresh(true)
}
func (b *Bucket) refresh(preserveConnections bool) error {
b.RLock()
pool := b.pool
uri := b.URI
client := pool.client
b.RUnlock()
var poolServices PoolServices
var err error
if client.tlsConfig != nil {
poolServices, err = client.GetPoolServices("default")
if err != nil {
return err
}
}
tmpb := &Bucket{}
err = pool.client.parseURLResponse(uri, tmpb)
if err != nil {
return err
}
pools := b.getConnPools(false /* bucket not already locked */)
// We need this lock to ensure that bucket refreshes happening because
// of NMVb errors received during bulkGet do not end up over-writing
// pool.inUse.
b.Lock()
for _, pool := range pools {
if pool != nil {
pool.inUse = false
}
}
newcps := make([]*connectionPool, len(tmpb.VBSMJson.ServerList))
for i := range newcps {
hostport := tmpb.VBSMJson.ServerList[i]
if preserveConnections {
pool := b.getConnPoolByHost(hostport, true /* bucket already locked */)
if pool != nil && pool.inUse == false && (!pool.encrypted || pool.tlsConfig == client.tlsConfig) {
// if the hostname and index is unchanged then reuse this pool
newcps[i] = pool
pool.inUse = true
continue
}
}
var encrypted bool
if client.tlsConfig != nil {
hostport, encrypted, err = MapKVtoSSL(hostport, &poolServices)
if err != nil {
b.Unlock()
return err
}
}
if b.ah != nil {
newcps[i] = newConnectionPool(hostport,
b.ah, AsynchronousCloser, PoolSize, PoolOverflow, client.tlsConfig, b.Name, encrypted)
} else {
newcps[i] = newConnectionPool(hostport,
b.authHandler(true /* bucket already locked */),
AsynchronousCloser, PoolSize, PoolOverflow, client.tlsConfig, b.Name, encrypted)
}
}
b.replaceConnPools2(newcps, true /* bucket already locked */)
tmpb.ah = b.ah
b.vBucketServerMap = unsafe.Pointer(&tmpb.VBSMJson)
b.nodeList = unsafe.Pointer(&tmpb.NodesJSON)
b.Unlock()
return nil
}
func (p *Pool) refresh() (err error) {
p.BucketMap = make(map[string]*Bucket)
buckets := []Bucket{}
err = p.client.parseURLResponse(p.BucketURL["uri"], &buckets)
if err != nil {
return err
}
for i, _ := range buckets {
b := new(Bucket)
*b = buckets[i]
b.pool = p
b.nodeList = unsafe.Pointer(&b.NodesJSON)
// MB-33185 this is merely defensive, just in case
// refresh() gets called on a perfectly node pool
ob, ok := p.BucketMap[b.Name]
if ok && ob.connPools != nil {
ob.Close()
}
b.replaceConnPools(make([]*connectionPool, len(b.VBSMJson.ServerList)))
p.BucketMap[b.Name] = b
runtime.SetFinalizer(b, bucketFinalizer)
}
buckets = nil
return nil
}
// GetPool gets a pool from within the couchbase cluster (usually
// "default").
func (c *Client) GetPool(name string) (p Pool, err error) {
var poolURI string
for _, p := range c.Info.Pools {
if p.Name == name {
poolURI = p.URI
break
}
}
if poolURI == "" {
return p, errors.New("No pool named " + name)
}
err = c.parseURLResponse(poolURI, &p)
if err != nil {
return p, err
}
p.client = c
err = p.refresh()
return
}
// GetPoolServices returns all the bucket-independent services in a pool.
// (See "Exposing services outside of bucket context" in http://goo.gl/uuXRkV)
func (c *Client) GetPoolServices(name string) (ps PoolServices, err error) {
var poolName string
for _, p := range c.Info.Pools {
if p.Name == name {
poolName = p.Name
}
}
if poolName == "" {
return ps, errors.New("No pool named " + name)
}
poolURI := "/pools/" + poolName + "/nodeServices"
err = c.parseURLResponse(poolURI, &ps)
return
}
func (b *Bucket) GetPoolServices(name string) (*PoolServices, error) {
b.RLock()
pool := b.pool
b.RUnlock()
ps, err := pool.client.GetPoolServices(name)
if err != nil {
return nil, err
}
return &ps, nil
}
// Close marks this bucket as no longer needed, closing connections it
// may have open.
func (b *Bucket) Close() {
b.Lock()
defer b.Unlock()
if b.connPools != nil {
for _, c := range b.getConnPools(true /* already locked */) {
if c != nil {
c.Close()
}
}
b.connPools = nil
}
}
func bucketFinalizer(b *Bucket) {
if b.connPools != nil {
if !b.closed {
logging.Warnf("Finalizing a bucket with active connections.")
}
// MB-33185 do not leak connection pools
b.Close()
}
}
// GetBucket gets a bucket from within this pool.
func (p *Pool) GetBucket(name string) (*Bucket, error) {
rv, ok := p.BucketMap[name]
if !ok {
return nil, &BucketNotFoundError{bucket: name}
}
err := rv.Refresh()
if err != nil {
return nil, err
}
return rv, nil
}
// GetBucket gets a bucket from within this pool.
func (p *Pool) GetBucketWithAuth(bucket, username, password string) (*Bucket, error) {
rv, ok := p.BucketMap[bucket]
if !ok {
return nil, &BucketNotFoundError{bucket: bucket}
}
rv.ah = newBucketAuth(username, password, bucket)
err := rv.Refresh()
if err != nil {
return nil, err
}
return rv, nil
}
// GetPool gets the pool to which this bucket belongs.
func (b *Bucket) GetPool() *Pool {
b.RLock()
defer b.RUnlock()
ret := b.pool
return ret
}
// GetClient gets the client from which we got this pool.
func (p *Pool) GetClient() *Client {
return p.client
}
// Release bucket connections when the pool is no longer in use
func (p *Pool) Close() {
// MB-36186 make the bucket map inaccessible
bucketMap := p.BucketMap
p.BucketMap = nil
// fine to loop through the buckets unlocked
// locking happens at the bucket level
for b, _ := range bucketMap {
// MB-36186 make the bucket unreachable and avoid concurrent read/write map panics
bucket := bucketMap[b]
bucketMap[b] = nil
bucket.Lock()
// MB-33208 defer closing connection pools until the bucket is no longer used
// MB-36186 if the bucket is unused make it unreachable straight away
needClose := bucket.connPools == nil && !bucket.closed
if needClose {
runtime.SetFinalizer(&bucket, nil)
}
bucket.closed = true
bucket.Unlock()
if needClose {
bucket.Close()
}
}
}
// GetBucket is a convenience function for getting a named bucket from
// a URL
func GetBucket(endpoint, poolname, bucketname string) (*Bucket, error) {
var err error
client, err := Connect(endpoint)
if err != nil {
return nil, err
}
pool, err := client.GetPool(poolname)
if err != nil {
return nil, err
}
return pool.GetBucket(bucketname)
}
// ConnectWithAuthAndGetBucket is a convenience function for
// getting a named bucket from a given URL and an auth callback
func ConnectWithAuthAndGetBucket(endpoint, poolname, bucketname string,
ah AuthHandler) (*Bucket, error) {
client, err := ConnectWithAuth(endpoint, ah)
if err != nil {
return nil, err
}
pool, err := client.GetPool(poolname)
if err != nil {
return nil, err
}
return pool.GetBucket(bucketname)
}
func GetSystemBucket(c *Client, p *Pool, name string) (*Bucket, error) {
bucket, err := p.GetBucket(name)
if err != nil {
if _, ok := err.(*BucketNotFoundError); !ok {
return nil, err
}
// create the bucket if not found
args := map[string]interface{}{
"authType": "sasl",
"bucketType": "couchbase",
"name": name,
"ramQuotaMB": 100,
"saslPassword": "donotuse",
}
var ret interface{}
// allow "bucket already exists" error in case duplicate create
// (e.g. two query nodes starting at same time)
err = c.parsePostURLResponseTerse("/pools/default/buckets", args, &ret)
if err != nil && !AlreadyExistsError(err) {
return nil, err
}
// bucket created asynchronously, try to get the bucket
maxRetry := 8
interval := 100 * time.Millisecond
for i := 0; i < maxRetry; i++ {
time.Sleep(interval)
interval *= 2
err = p.refresh()
if err != nil {
return nil, err
}
bucket, err = p.GetBucket(name)
if bucket != nil {
bucket.RLock()
ok := !bucket.closed && len(bucket.getConnPools(true /* already locked */)) > 0
bucket.RUnlock()
if ok {
break
}
} else if err != nil {
if _, ok := err.(*BucketNotFoundError); !ok {
break
}
}
}
}
return bucket, err
}
func DropSystemBucket(c *Client, name string) error {
err := c.parseDeleteURLResponseTerse("/pools/default/buckets/"+name, nil, nil)
return err
}
func AlreadyExistsError(err error) bool {
// Bucket error: Bucket with given name already exists
// Scope error: Scope with this name already exists
// Collection error: Collection with this name already exists
return strings.Contains(err.Error(), " name already exists")
}